修改机器人选股现金流筛选标准，自由设定经营现金流下限阀值。

机器人选股现金流筛选自由设定经营现金流下限阈值一、实际应用场景描述在量化选股策略中经营现金流Operating Cash Flow, OCF 是衡量企业造血能力最核心的指标之一。相比净利润经营现金流更难以被会计手段粉饰——净利润可以调节但银行里的真金白银骗不了人。典型场景场景一利润增长但现金流恶化某消费股 2023 年报显示净利润同比增长 15%看起来健康。但细看经营现金流从 5 亿骤降至 -2 亿。原因是应收账款激增、存货积压。半年后公司爆出债务危机股价腰斩。如果策略在选股时过滤掉 OCF 为负或低于阈值的标的就能避开这颗雷。场景二周期性行业现金流波动某钢铁企业在行业下行期经营现金流从 20 亿降至 3 亿虽仍为正但大幅收缩。策略若设定 OCF 下限为近四季均 5 亿可提前过滤掉这类衰退期标的。场景三跨行业对比失真直接比较不同行业公司的 OCF 绝对值无意义银行天生现金流巨大软件公司则小。因此需要行业中性化处理——将 OCF 除以营业收入或市值得到现金流收益率再比较。二、引入痛点痛点 具体表现 利润 ≠ 真金白银 大量公司净利润好看经营现金流持续为负财务造假高发区 跨行业不可比 银行 OCF 百亿 vs 软件公司 OCF 千万绝对值比较无意义 单期数据噪声大 某季度 OCF 为负可能是季节性如零售 Q4 备货单期筛选误杀 阈值拍脑袋 现金流 0太宽松 5 亿太严格缺乏系统化参数扫描 财报发布滞后 最新财报可能滞后 3 个月选股时用到的是过期数据 非经常性损益干扰 一次性资产处置带来的现金流激增不是持续经营能力三、核心逻辑讲解3.1 为什么是经营现金流OCF三大现金流对比┌─────────────────────────────────────────────────────┐│ 类型 │ 反映什么 │ 可调节性 │├─────────────────────┼────────────────────┼───────────┤│ 经营活动现金流 OCF │ 主营业务造血能力 │ ★☆☆ 最难 ││ 投资活动现金流 ICF │ 扩张/收缩姿态 │ ★★☆ 中等 ││ 筹资活动现金流 FCF │ 融资/分红/回购 │ ★★★ 最易 │└─────────────────────────────────────────────────────┘OCF 净利润 折旧摊销 营运资本变动调整核心逻辑一家公司可以没有利润战略亏损但不能没有经营现金流——否则连工资都发不出离退市不远了。3.2 现金流筛选维度设计维度 指标 说明绝对值 OCF 最近季度 阈值 最基础过滤相对值 OCF / 营业收入 阈值 行业中性化消除规模差异趋势 OCF 近 4 季环比变化 阈值 捕捉现金流恶化/改善趋势稳定性 OCF 近 4 季标准差 / 均值 阈值 排除波动过大的不可靠现金流持续性 连续 N 季 OCF 0 排除偶尔为正的偶然情况3.3 筛选逻辑流程┌──────────────────────────────────────────────────────────┐│ AI 选股 → 现金流筛选模块 │├──────────────────────────────────────────────────────────┤│ ││ 候选股票池AI 打分后 ││ │ ││ ▼ ││ ┌──────────────────────────────────┐ ││ │ ★ 第一关OCF 绝对值阈值 │ ││ │ 最近季度 OCF min_ocf_abs │ ← 如 1 亿 ││ └──────────────────────────────────┘ ││ │ 通过 ││ ▼ ││ ┌──────────────────────────────────┐ ││ │ ★ 第二关OCF/营收 相对阈值 │ ││ │ OCF / Revenue min_ocf_rel │ ← 如 5% ││ └──────────────────────────────────┘ ││ │ 通过 ││ ▼ ││ ┌──────────────────────────────────┐ ││ │ ★ 第三关趋势检查 │ ││ │ 近 4 季 OCF 无连续下降 │ ││ │ 可选防恶化 │ ││ └──────────────────────────────────┘ ││ │ 通过 ││ ▼ ││ 最终入选标的 ││ │└──────────────────────────────────────────────────────────┘3.4 阈值设定参考A 股经验值参数 宽松成长策略 中性均衡策略 严格价值策略min_ocf_abs亿 0.5 1.0 5.0min_ocf_revenue_pct 3% 5% 10%min_consecutive_positive 2 季 3 季 4 季max_ocf_decline_pct -30% -20% -10%四、项目结构cashflow_filter/├── README.md├── requirements.txt├── config.yaml # 全局配置含现金流阈值├── data/│ ├── stock_universe.csv # 股票池含基本面数据│ └── sample_cashflow.csv # 季度现金流数据├── src/│ ├── data_loader.py # 数据加载含现金流│ ├── cashflow_filter.py # ★ 现金流筛选核心模块│ ├── cashflow_analyzer.py # 现金流深度分析趋势/稳定性│ ├── stock_scorer.py # AI 选股打分│ ├── strategy_engine.py # 策略引擎集成现金流过滤│ ├── backtester.py # 回测框架│ ├── sensitivity_scanner.py # ★ 阈值敏感性扫描│ └── visualizer.py # 可视化├── main.py # 主入口└── compare_cashflow_thresholds.py # 不同阈值对比实验五、完整代码模块化 清晰注释requirements.txtpandas1.5numpy1.21matplotlib3.5seaborn0.12scipy1.9pyyaml6.0config.yaml# 现金流筛选配置# ★ 现金流筛选阈值可自由设定cashflow:enabled: true# 绝对值阈值 min_ocf_abs: 1.0 # 最近季度 OCF 1 亿亿min_ocf_ttm: 3.0 # 近四季滚动 OCF 3 亿亿# 相对值阈值行业中性min_ocf_revenue_pct: 0.05 # OCF / 营业收入 5%min_ocf_market_cap_pct: 0.02 # OCF / 市值 2%现金流收益率# 趋势检查 min_consecutive_positive: 3 # 连续 3 个季度 OCF 0max_ocf_decline_pct: -0.20 # 近两季 OCF 降幅不超过 -20%# 稳定性检查 max_ocf_cv: 0.50 # OCF 变异系数标准差/均值 50%# 数据来源cashflow_period: quarterly # quarterly / annuallookback_quarters: 8 # 回溯 8 个季度# 策略参数strategy:initial_capital: 10000000max_positions: 10take_profit_pct: 0.08stop_loss_pct: -0.05commission_rate: 0.0003stamp_tax_rate: 0.001model:features:- momentum_20d- ocf_yield # ★ 现金流收益率作为特征- ocf_trend # ★ 现金流趋势作为特征- revenue_growthscore_threshold: 0.5src/data_loader.pydata_loader.py数据加载模块股票池 季度现金流数据import pandas as pdfrom pathlib import Pathfrom typing import Optional, Tupledef load_stock_universe(filepath: str) - pd.DataFrame:加载股票池含市值、营收等基本面数据预期 CSV 格式:code,name,industry,market_cap(billion),revenue_ttm(billion)000001,平安银行,金融,231.5,156.8600519,贵州茅台,消费,2180.0,147.7...df pd.read_csv(filepath, parse_dates[list_date])df[code] df[code].astype(str).str.zfill(6)# 确保数值列为 numericfor col in [market_cap(billion), revenue_ttm(billion)]:if col in df.columns:df[col] pd.to_numeric(df[col], errorscoerce)return dfdef load_cashflow_data(filepath: str) - pd.DataFrame:加载季度现金流数据预期 CSV 格式:code,report_date,ocf_q,revenue_q,net_profit_q000001,2024-Q3,12.5,45.2,10.8000001,2024-Q2,10.3,43.1,9.5...其中 ocf_q 当季经营活动现金流净额亿元df pd.read_csv(filepath, parse_dates[report_date])df[code] df[code].astype(str).str.zfill(6)# 确保数值列for col in [ocf_q, revenue_q, net_profit_q]:if col in df.columns:df[col] pd.to_numeric(df[col], errorscoerce)# 按股票和报告期排序df df.sort_values([code, report_date])return dfdef generate_mock_cashflow_data(stock_universe: pd.DataFrame,n_quarters: int 12,seed: int 42) - pd.DataFrame:为股票池生成模拟季度现金流数据模拟逻辑- 大市值公司 → 更大的 OCF 绝对值- OCF / Revenue 比率在合理区间5%~25%- 引入部分现金流恶化的标的约 20%import numpy as npfrom dateutil.relativedelta import relativedeltanp.random.seed(seed)records []base_date pd.Timestamp(2022-03-31)for _, stock in stock_universe.iterrows():code stock[code]mcap stock.get(market_cap(billion), 10.0)revenue stock.get(revenue_ttm(billion), mcap * 0.5)# 基础 OCF与市值和营收正相关base_ocf revenue * np.random.uniform(0.05, 0.20)# 20% 概率生成现金流恶化标的is_deteriorating np.random.random() 0.20for q in range(n_quarters):report_date base_date relativedelta(months3 * q)# 季节性波动seasonal 1.0 0.15 * np.sin(2 * np.pi * q / 4)if is_deteriorating:# 现金流持续恶化deterioration 1.0 - 0.08 * qocf base_ocf * max(deterioration, 0.1) * seasonalif q 4:ocf * 0.5 # 加速恶化else:# 正常波动 缓慢增长growth 1.0 0.02 * qnoise np.random.normal(1.0, 0.15)ocf base_ocf * growth * noise * seasonalocf max(ocf, -base_ocf * 0.3) # 允许偶尔为负q_revenue revenue / 4 * np.random.uniform(0.8, 1.2)q_profit q_revenue * np.random.uniform(0.05, 0.20)records.append({code: code,report_date: report_date,ocf_q: round(ocf, 2),revenue_q: round(q_revenue, 2),net_profit_q: round(q_profit, 2)})return pd.DataFrame(records)def get_close_price_matrix(price_data: pd.DataFrame) - pd.DataFrame:收盘价矩阵return price_data[close].unstack()src/cashflow_filter.py★ 核心模块cashflow_filter.py★ 经营现金流筛选模块多维度过滤造血能力不达标标的核心功能:1. OCF 绝对值门槛亿2. OCF / 营收 相对门槛行业中性3. OCF / 市值 现金流收益率4. 连续为正检查5. 趋势恶化检查6. 稳定性变异系数检查import pandas as pdimport numpy as npfrom typing import Dict, List, Optional, Tupleimport logginglogging.basicConfig(levellogging.INFO, format%(asctime)s [%(levelname)s] %(message)s)logger logging.getLogger(__name__)class CashflowFilter:经营现金流筛选器在 AI 模型打分之后、实际买入之前对候选标的进行现金流体检。只有通过了所有门槛的标的才允许进入最终买入列表。def __init__(self,# 绝对值阈值min_ocf_abs: float 1.0, # 最近季度 OCF 1 亿min_ocf_ttm: float 3.0, # 近四季 OCF 3 亿# 相对值阈值行业中性min_ocf_revenue_pct: float 0.05, # OCF/营收 5%min_ocf_market_cap_pct: float 0.02, # OCF/市值 2%# 趋势检查min_consecutive_positive: int 3, # 连续 3 季为正max_ocf_decline_pct: float -0.20, # 降幅不超过 -20%# 稳定性max_ocf_cv: float 0.50, # 变异系数 50%# 数据参数lookback_quarters: int 8 # 回溯季度数):参数说明:min_ocf_abs: 最近单季度 OCF 绝对值下限亿元min_ocf_ttm: 近四季滚动 OCF 下限亿元min_ocf_revenue_pct: OCF 占营收比例下限如 0.05 5%min_ocf_market_cap_pct: OCF 占市值比例下限现金流收益率min_consecutive_positive: 连续为正的最少季度数max_ocf_decline_pct: 相邻两季 OCF 最大允许降幅如 -0.20 -20%max_ocf_cv: OCF 变异系数上限标准差/均值lookback_quarters: 分析回溯的季度数self.min_ocf_abs min_ocf_absself.min_ocf_ttm min_ocf_ttmself.min_ocf_revenue_pct min_ocf_revenue_pctself.min_ocf_market_cap_pct min_ocf_market_cap_pctself.min_consec_positive min_consecutive_positiveself.max_ocf_decline max_ocf_decline_pctself.max_ocf_cv max_ocf_cvself.lookback lookback_quarterslogger.info(f现金流筛选器初始化:)logger.info(f 绝对值: 单季{min_ocf_abs}亿, TTM{min_ocf_ttm}亿)logger.info(f 相对值: OCF/营收{min_ocf_revenue_pct*100:.0f}%, fOCF/市值{min_ocf_market_cap_pct*100:.0f}%)logger.info(f 趋势: 连续{min_consec_positive}季为正, 最大降幅{max_ocf_decline_pct*100:.0f}%)logger.info(f 稳定性: CV{max_ocf_cv*100:.0f}%)def filter(self,candidates: pd.DataFrame,cashflow_data: pd.DataFrame,reference_date: pd.Timestamp,stock_universe: Optional[pd.DataFrame] None) - Tuple[pd.DataFrame, Dict]:★ 核心方法对候选股票执行现金流筛选参数:candidates: AI 选股后的候选池需含 code 列cashflow_data: 季度现金流数据reference_date: 当前参考日期决定用哪期财报stock_universe: 股票基本面数据含市值、营收 TTM返回:Tuple[过滤后股票池, 详细筛选报告]original_count len(candidates)if len(candidates) 0:return candidates, {original_count: 0, pass_count: 0, rejections: {}}# 确定每只股票最新的财报期latest_reports self._get_latest_reports(cashflow_data, candidates[code].tolist(), reference_date)# 合并最新财报数据cf_latest cashflow_data[cashflow_data.set_index([code, report_date]).index.isin(latest_reports)].copy()# 合并历史数据用于趋势分析cf_history self._get_historical_data(cashflow_data, candidates[code].tolist(), reference_date)# 合并市值和营收数据 if stock_universe is not None:fundamentals stock_universe.set_index(code)[[market_cap(billion), revenue_ttm(billion)]].copy()fundamentals.columns [market_cap_b, revenue_ttm_b]cf_latest cf_latest.merge(fundamentals, oncode, howleft)# 执行五层筛选 results pd.DataFrame({code: candidates[code]})results[passed] Trueresults[reject_reason] results results.set_index(code)rejections {} # 统计各关卡拒绝数# ★ 关卡 1: OCF 绝对值passed_1, reason_1 self._check_absolute_ocf(cf_latest)results self._apply_filter(results, passed_1, reason_1, rejections)# ★ 关卡 2: OCF / 营收passed_2, reason_2 self._check_ocf_revenue_ratio(cf_latest)results self._apply_filter(results, passed_2, reason_2, rejections)# ★ 关卡 3: OCF / 市值现金流收益率passed_3, reason_3 self._check_ocf_market_cap_ratio(cf_latest)results self._apply_filter(results, passed_3, reason_3, rejections)# ★ 关卡 4: 连续为正passed_4, reason_4 self._check_consecutive_positive(cf_history)results self._apply_filter(results, passed_4, reason_4, rejections)# ★ 关卡 5: 趋势恶化检查passed_5, reason_5 self._check_trend(cf_history)results self._apply_filter(results, passed_5, reason_5, rejections)# ★ 关卡 6: 稳定性变异系数passed_6, reason_6 self._check_stability(cf_history)results self._apply_filter(results, passed_6, reason_6, rejections)# 汇总 final results[results[passed]].index.tolist()filtered candidates[candidates[code].isin(final)]pass_count len(filtered)removed_count original_count - pass_countlogger.info(f[{reference_date.strftime(%Y-%m-%d)}] f现金流筛选: {original_count} → {pass_count} f(移除 {removed_count}, {removed_count/max(original_count,1)*100:.1f}%))if rejections:logger.debug( 各关卡拒绝统计:)for reason, count in sorted(rejections.items(), keylambda x: -x[1]):logger.debug(f {reason}: {count} 只)report {original_count: original_count,pass_count: pass_count,removed_count: removed_count,rejections: rejections,pass_rate_pct: pass_count / max(original_count, 1) * 100}return filtered, reportdef _get_latest_reports(self,cf_data: pd.DataFrame,codes: List[str],ref_date: pd.Timestamp) - List[Tuple[str, pd.Timestamp]]:获取每只股票在参考日期前的最新一期财报latest []for code in codes:mask (cf_data[code] code) (cf_data[report_date] ref_date)stock_cf cf_data[mask].sort_values(report_date, ascendingFalse)if len(stock_cf) 0:latest.append((code, stock_cf.iloc[0][report_date]))return latestdef _get_historical_data(self,cf_data: pd.DataFrame,codes: List[str],ref_date: pd.Timestamp) - pd.DataFrame:获取回溯期内的历史现金流数据mask ((cf_data[code].isin(codes)) (cf_data[report_date] ref_date) (cf_data[report_date] ref_date - pd.Timedelta(daysself.lookback * 100)))return cf_data[mask].copy()def _apply_filter(self,results: pd.DataFrame,passed: pd.Series,reason: str,rejections: Dict) - pd.DataFrame:将单关卡结果应用到总表mask results[passed] passedrejected results[(results[passed]) (~passed)]results.loc[~mask, passed] Falseresults.loc[rejected.index, reject_reason] reasonreject_count (~passed).sum()if reject_count 0:rejections[reason] rejections.get(reason, 0) int(reject_count)return resultsdef _check_absolute_ocf(self, cf: pd.DataFrame) - Tuple[pd.Series, str]:关卡 1: OCF 绝对值 TTM 检查# 单季度检查mask_q cf[ocf_q].fillna(-9999) self.min_ocf_abs# TTM近四季滚动检查if ocf_ttm in cf.columns:mask_ttm cf[ocf_ttm].fillna(-9999) self.min_ocf_ttmelse:mask_ttm pd.Series(True, indexcf.index)return mask_q mask_ttm, fOCF_abs{self.min_ocf_abs}亿def _check_ocf_revenue_ratio(self, cf: pd.DataFrame) - Tuple[pd.Series, str]:关卡 2: OCF / 营收 相对阈值if revenue_q not in cf.columns or revenue_ttm_b not in cf.columns:return pd.Series(True, indexcf.index), # 用季度营收或 TTM 营收优先 TTMrevenue cf[revenue_ttm_b].fillna(cf[revenue_q])ratio cf[ocf_q] / revenue.replace(0, float(nan))return ratio.fillna(-9999) self.min_ocf_revenue_pct, \fOCF/营收{self.min_ocf_revenue_pct*100:.0f}%def _check_ocf_market_cap_ratio(self, cf: pd.DataFrame) - Tuple[pd.Series, str]:关卡 3: OCF / 市值现金流收益率if market_cap_b not in cf.columns:return pd.Series(True, indexcf.index), # 将 OCF 从亿转为与市值同单位或直接计算比例# OCF 是季度值亿市值也是亿ratio cf[ocf_q] / cf[market_cap_b].replace(0, float(nan))return ratio.fillna(-9999) self.min_ocf_market_cap_pct, \fOCF/市值{self.min_ocf_market_cap_pct*100:.0f}%def _check_consecutive_positive(self, cf_hist: pd.DataFrame) - Tuple[pd.Series, str]:关卡 4: 连续 N 季 OCF 0if len(cf_hist) 0:return pd.Series(True, indexcf_hist.index[:0]), results {}for code, group in cf_hist.groupby(code):ocf_series group.sort_values(report_date)[ocf_q].values# 取最后 N 个季度recent ocf_series[-self.min_consec_positive:]all_positive all(v 0 for v in recent)results[code] all_positivereturn pd.Series(results), f连续{self.min_consec_positive}季OCF未全为正def _check_trend(self, cf_hist: pd.DataFrame) - Tuple[pd.Series, str]:关卡 5: 趋势恶化检查相邻季度降幅if len(cf_hist) 0:return pd.Series(True, indexcf_hist.index[:0]), results {}for code, group in cf_hist.groupby(code):ocf_sorted group.sort_values(report_date)[ocf_q].valuesif len(ocf_sorted) 2:results[code] Truecontinue# 检查最近两季的降幅latest ocf_sorted[-1]prev ocf_sorted[-2]if prev 0:decline (latest - prev) / prevresults[code] decline self.max_ocf_declineelse:# 前期已经为负不额外惩罚results[code] Truereturn pd.Series(results), fOCF降幅{abs(self.max_ocf_decline)*100:.0f}%def _check_stability(self, cf_hist: pd.DataFrame) - Tuple[pd.Series, str]:关卡 6: 稳定性变异系数 CV std/meanif len(cf_hist) 0:return pd.Series(True, indexcf_hist.index[:0]), results {}for code, group in cf_hist.groupby(code):ocf_values group[ocf_q].valuesif len(ocf_values) 3:results[code] Truecontinue本文代码仅供学习与技术交流不构成任何投资建议股市有风险入市需谨慎利用AI解决实际问题如果你觉得这个工具好用欢迎关注长安牧笛