聚焦“通用趋势策略 成交额流动性过滤”这一件事适合直接写进课程讲义或技术博客。通用趋势策略增加流动性过滤剔除成交额过低个股一、实际应用场景描述趋势策略Trend Following是量化投资中最常见、最“通用”的策略骨架之一均线金叉 → 做多均线死叉 → 止损/做空。但在实盘落地时一个被广泛忽视、却频繁“吃人”的问题就是 成交额过低带来的隐性风险真实场景 问题本质策略选出一只“技术形态完美”的小盘股 日均成交额只有 300 万计划买入 50 万挂单后发现 冲击成本高达 3%~5%持有期间遇到利空 想卖但盘口没有承接连续跌停回测中“完美止盈” 实盘直接变成“止盈变巨亏” 回测里你是用收盘价“无摩擦”成交的而现实是没有流动性就没有执行力。二、引入痛点问题结构化我们把问题拆成可被工程化解决的几个层级层级 痛点 后果数据层 只看价格/均线忽略成交额 选到“纸面趋势”策略层 没有流动性约束 实盘无法复现回测风控层 缺乏“最坏情况可退出”的保证 单次黑天鹅致命系统工程层 参数拍脑袋 不知道 3000 万还是 1 亿更合适教学层 趋势策略 demo 过度简化 学生以为“均线好用”本质结论没有流动性约束的趋势策略不是策略是假设。三、核心逻辑讲解“为什么要这么设计”3.1 成交额为什么比市值更“真实”很多教程喜欢用市值来过滤流动性但在 A 股- 大量小盘股 市值小 ≠ 成交额小- 也存在 大盘股流动性崩塌 的情况如停牌复盘、利空连续一字板✅ 成交额Turnover / Amount 是- 实时可观测- 可直接回测- 可直接映射到“能否按计划成交”3.2 流动性过滤的“三层结构”我们引入一个分层流动性过滤框架┌─────────────────────────────────────────────┐│ 成交额流动性过滤三层 │├─────────────────────────────────────────────┤│ ││ Layer 1绝对成交额门槛 ││ ───────────────────────────────────── ││ 近 N 日日均成交额 ≥ 阈值如 2000 万 ││ 过滤“僵尸股” ││ ││ Layer 2相对成交额稳定性 ││ ───────────────────────────────────── ││ 近 N 日成交额标准差 / 均值 ≤ 阈值 ││ 过滤“偶尔放量、平时没流动性”的个股 ││ ││ Layer 3极端低流动性预警 ││ ───────────────────────────────────── ││ 当日成交额 阈值 → 强制不交易 ││ 防止“突然失去流动性”的尾部风险 ││ │└─────────────────────────────────────────────┘3.3 核心参数设计可解释、可回测参数 含义 经验值A 股min_avg_amount 日均成交额下限 1000 万 ~ 5000 万lookback_days 计算窗口 20 日max_cv 变异系数上限 ≤ 0.8emergency_amount 当日熔断阈值 300 万min_trading_days 最少交易日 ≥ 15 天3.4 和趋势策略如何“正确集成”✅ 正确顺序这是关键工程细节选股信号生成↓趋势策略筛选均线 / 突破 / 动量↓★ 流动性过滤先粗筛 → 再精筛↓下单执行❌ 常见错误- 先过滤流动性 → 再算趋势破坏了策略逻辑- 在回测里用 future data用未来成交额决定过去是否交易四、项目结构工程化trend_liquidity_filter/├── README.md├── requirements.txt├── config.yaml├── data/│ ├── daily_prices.csv # 日频行情│ └── daily_amounts.csv # 日频成交额├── src/│ ├── data_loader.py # 数据加载│ ├── trend_signal.py # 通用趋势策略均线/突破│ ├── liquidity_filter.py # ★ 流动性过滤核心模块│ ├── strategy_engine.py # 策略引擎│ ├── backtester.py # 回测框架│ └── visualizer.py # 可视化├── main.py└── compare_liquidity.py # 有/无流动性过滤对比五、完整代码模块化 清晰注释requirements.txtpandas1.5numpy1.21matplotlib3.5seaborn0.12pyyaml6.0config.yaml# 趋势策略 流动性过滤配置# 流动性过滤参数liquidity:enabled: truelookback_days: 20min_avg_amount: 20000000 # 2000 万单位元max_cv: 0.8 # 变异系数上限emergency_amount: 3000000 # 单日 300 万 → 禁止交易min_trading_days: 15# 趋势策略参数strategy:fast_ma: 5slow_ma: 20max_positions: 5take_profit_pct: 0.08stop_loss_pct: -0.05initial_capital: 1000000commission_rate: 0.0003stamp_tax_rate: 0.001backtest:start_date: 2022-01-01end_date: 2024-12-31src/data_loader.pydata_loader.py行情 成交额数据加载import pandas as pddef load_price_data(path: str) - pd.DataFrame:CSV 格式:date,code,open,high,low,closedf pd.read_csv(path, parse_dates[date])df[code] df[code].astype(str).str.zfill(6)return df.set_index([date, code]).sort_index()def load_amount_data(path: str) - pd.DataFrame:CSV 格式:date,code,amountdf pd.read_csv(path, parse_dates[date])df[code] df[code].astype(str).str.zfill(6)return df.set_index([date, code]).sort_index()def get_close_matrix(price_data: pd.DataFrame) - pd.DataFrame:return price_data[close].unstack()def get_amount_matrix(amount_data: pd.DataFrame) - pd.DataFrame:return amount_data[amount].unstack()src/trend_signal.pytrend_signal.py通用趋势策略双均线 突破import pandas as pdimport numpy as npdef compute_trend_signals(close: pd.DataFrame,fast: int 5,slow: int 20) - pd.DataFrame:为每只股票计算趋势信号返回 DataFrame:- ma_fast / ma_slow / trend_signal- trend_signal: 1 做多, 0 无信号signals pd.DataFrame(indexclose.index, columnsclose.columns)for code in close.columns:s close[code].dropna()if len(s) slow:continuema_fast s.rolling(fast).mean()ma_slow s.rolling(slow).mean()# ★ 核心趋势逻辑signal (ma_fast ma_slow).astype(int)signals[code] signalreturn signalsdef select_trend_candidates(signals: pd.DataFrame,date: pd.Timestamp,n_candidates: int 20) - list:选择当日处于“上升趋势”的股票if date not in signals.index:return []row signals.loc[date]candidates row[row 1].index.tolist()return candidates[:n_candidates]src/liquidity_filter.py★ 核心模块liquidity_filter.py★ 成交额流动性过滤三层结构import pandas as pdimport numpy as npfrom typing import Dict, Listclass LiquidityFilter:成交额流动性过滤器三层过滤1. 日均成交额 ≥ min_avg_amount2. 成交额变异系数 ≤ max_cv3. 当日成交额 ≥ emergency_amountdef __init__(self,lookback_days: int 20,min_avg_amount: float 20_000_000,max_cv: float 0.8,emergency_amount: float 3_000_000,min_trading_days: int 15):self.lookback lookback_daysself.min_avg min_avg_amountself.max_cv max_cvself.emergency emergency_amountself.min_days min_trading_daysdef filter(self,candidates: List[str],amount_matrix: pd.DataFrame,date: pd.Timestamp) - Dict:★ 核心方法返回:{passed: [code, ...],rejected: {avg_amount: [code, ...],cv: [code, ...],emergency: [code, ...],insufficient_data: [code, ...]}}passed []rejected {avg_amount: [],cv: [],emergency: [],insufficient_data: []}# 计算窗口start date - pd.Timedelta(daysself.lookback * 1.5)window amount_matrix.loc[start:date]for code in candidates:if code not in amount_matrix.columns:rejected[insufficient_data].append(code)continueseries window[code].dropna()# ★ Layer 0数据充足性检查if len(series) self.min_days:rejected[insufficient_data].append(code)continue# ★ Layer 1日均成交额avg series.mean()if avg self.min_avg:rejected[avg_amount].append(code)continue# ★ Layer 2成交额稳定性变异系数cv series.std() / avg if avg 0 else float(inf)if cv self.max_cv:rejected[cv].append(code)continue# ★ Layer 3当日成交额熔断today_amount series.iloc[-1] if len(series) 0 else 0if today_amount self.emergency:rejected[emergency].append(code)continuepassed.append(code)return {passed: passed,rejected: rejected}def print_report(self, rejected: Dict, date: pd.Timestamp):total_rejected sum(len(v) for v in rejected.values())print(f [{date.strftime(%Y-%m-%d)}] 流动性过滤:)print(f 候选: {sum(len(v) for v in rejected.values()) len(self.passed)})for reason, codes in rejected.items():if codes:print(f {reason}: {len(codes)} 只)src/strategy_engine.pystrategy_engine.py趋势策略引擎集成流动性过滤import pandas as pdimport numpy as npfrom src.liquidity_filter import LiquidityFilterclass TrendLiquidityStrategy:★ 趋势策略 流动性过滤def __init__(self,liquidity_filter: LiquidityFilter,fast_ma: int 5,slow_ma: int 20,max_positions: int 5,take_profit_pct: float 0.08,stop_loss_pct: float -0.05,initial_capital: float 1_000_000,commission_rate: float 0.0003,stamp_tax_rate: float 0.001):self.lf liquidity_filterself.fast fast_maself.slow slow_maself.max_pos max_positionsself.tp take_profit_pctself.sl stop_loss_pctself.comm commission_rateself.tax stamp_tax_rateself.capital initial_capitalself.positions {}self.daily_nav {}self.trade_log []self.total_rejected 0print(f\n{*60})print(f 趋势策略 流动性过滤引擎)print(f 均线: {fast_ma}/{slow_ma})print(f 流动性门槛: ¥{liquidity_filter.min_avg/1e4:.0f}万/日)print(f{*60}\n)def run_daily(self,date: pd.Timestamp,close: pd.Series,amount_matrix: pd.DataFrame,trend_signals: pd.DataFrame):# 1) 生成趋势候选candidates self._get_trend_candidates(trend_signals, date)# ★ 2) 流动性过滤result self.lf.filter(candidates, amount_matrix, date)passed result[passed]self.total_rejected sum(len(v) for v in result[rejected].values())# 3) 买入if len(self.positions) self.max_pos:for code in passed:if len(self.positions) self.max_pos:breakif code in self.positions:continueif code not in close or close[code] 0:continueself._open(code, date, close[code])# 4) 止盈止损self._check_exit(date, close)# 5) 记录净值self._record_nav(date, close)def _get_trend_candidates(self, signals, date) - list:if date not in signals.index:return []row signals.loc[date]return row[row 1].index.tolist()def _open(self, code, date, price):alloc self.capital / max(1, self.max_pos - len(self.positions))qty int(alloc / price / 100) * 100if qty 0:returncost qty * price * (1 self.comm)if cost self.capital:returnself.capital - costself.positions[code] {open_date: date,open_price: price,qty: qty}def _check_exit(self, date, prices):to_close []for code, pos in self.positions.items():if code not in prices or prices[code] 0:continuepnl (prices[code] - pos[open_price]) / pos[open_price]if pnl self.tp or pnl self.sl:to_close.append(code)for code in to_close:px prices.get(code, self.positions[code][open_price])self._close(code, date, px)def _close(self, code, date, price):pos self.positions.pop(code, None)if not pos:returnrevenue pos[qty] * price * (1 - self.comm - self.tax)self.capital revenuedef _record_nav(self, date, prices):nav self.capitalfor code, pos in self.positions.items():if code in prices and prices[code] 0:nav pos[qty] * prices[code]self.daily_nav[date] navdef final_liquidate(self, date, prices):for code in list(self.positions.keys()):px prices.get(code, self.positions[code][open_price])self._close(code, date, px)def print_stats(self):print(f\n{*60})print(f 期末资金: ¥{self.capital:,.2f})print(f 累计拦截: {self.total_rejected} 次)print(f{*60}\n)src/backtester.pybacktester.py回测引擎含对比模式import pandas as pdfrom src.strategy_engine import TrendLiquidityStrategydef run_backtest(strategy: TrendLiquidityStrategy,close: pd.DataFrame,amount_matrix: pd.DataFrame,trend_signals: pd.DataFrame,start: str None,end: str None,enable_liquidity: bool True) - dict:dates close.indexif start:dates dates[dates pd.Timestamp(start)]if end:dates dates[dates pd.Timestamp(end)]for i, dt in enumerate(dates):day_close close.loc[dt] if dt in close.index else pd.Series()strategy.run_daily(dt, day_close, amount_matrix, trend_signals)if i % max(1, len(dates)//10) 0:nav strategy.daily_nav.get(dt, 0)print(f [{i1}/{len(dates)}] {dt.strftime(%Y-%m-%d)} | f持仓:{len(strategy.positions)} | 净值:¥{nav:,.0f})last dates[-1]last_px close.loc[last] if last in close.index else pd.Series()strategy.final_liquidate(last, last_px)return {nav: pd.Series(strategy.daily_nav), strategy: strategy}def calc_metrics(nav: pd.Series) - dict:nav nav.dropna()if len(nav) 2:return {}ret nav.pct_change().dropna()days (nav.index[-1] - nav.index[0]).daysyrs days / 365.25tot nav.iloc[-1]/nav.iloc[0] - 1ann (1tot)**(1/yrs) - 1 if yrs 0 else 0vol ret.std() * 252**0.5sharpe (ann - 0.025)/vol if vol 0 else 0dd (nav - nav.cummax())/nav.cummax()return {total_ret_pct: round(tot*100, 2),ann_ret_pct: round(ann*100, 2),vol_pct: round(vol*100, 2),sharpe: round(sharpe, 3),max_dd_pct: round(dd.min()*100, 2),final_equity: round(nav.iloc[-1], 2),n_rejected: strategy.total_rejected if strategy in locals() else 0}def print_comparison(m1: dict, m2: dict,l1含流动性过滤, l2无流动性过滤):print(f\n{*65})print(f{指标:22}{l1:18}{l2:18}{差异:10})print(f{*65})for lab, key, fmt in [(累计收益(%),total_ret_pct,.2f),(年化(%),ann_ret_pct,.2f),(波动(%),vol_pct,.2f),(夏普,sharpe,.3f),(最大回撤(%),max_dd_pct,.2f),(期末权益,final_equity,.0f)]:v1, v2 m1.get(key,0), m2.get(key,0)fs f{{:22}}{{:18{fmt}}}{{:18{fmt}}}{{:15{fmt}}}print(fs.format(lab, v1, v2, v1-v2))print(f{*65}\n)main.pymain.py趋势策略 流动性过滤回测import yamlimport pandas as pdimport numpy as npfrom src.data_loader import load_price_data, load_amount_data, get_close_matrix, get_amount_matrixfrom src.trend_signal import compute_trend_signals, select_trend_candidatesfrom src.liquidity_filter import LiquidityFilterfrom src.strategy_engine import TrendLiquidityStrategyfrom src.backtester import run_backtest, calc_metrics, print_comparisondef load_config(pathconfig.yaml):with open(path) as f:return yaml.safe_load(f)def generate_mock_data(n_stocks30, seed42):np.random.seed(seed)dates pd.date_range(2022-01-01, 2024-12-31, freqB)codes [f{i:06d} for i in range(n_stocks)]price_recs []amount_recs []for code in codes:drift np.random.normal(0.0003, 0.015, len(dates))close 10 * np.cumprod(1 drift)for d, c in zip(dates, close):price_recs.append({date: d, code: code, close: round(c, 2)})# 成交额大部分有流动性部分模拟僵尸股if np.random.random() 0.2:amt np.random.uniform(1e5, 5e5) # 低流动性else:amt np.random.uniform(2e6, 5e7) # 正常amount_recs.append({date: d, code: code, amount: round(amt, 2)})return (pd.DataFrame(price_recs), pd.DataFrame(amount_recs))def main():cfg load_config()# 数据try:prices load_price_data(data/daily_prices.csv)amounts load_amount_data(data/daily_amounts.csv)close get_close_matrix(prices)amount_mtx get_amount_matrix(amounts)except FileNotFoundError:print(生成模拟数据...)p, a generate_mock_data()close p.pivot(indexdate, columnscode, valuesclose)amount_mtx a.pivot(indexdate, columnscode, valuesamount)# 趋势信号signals compute_trend_signals(close, cfg[strategy][fast_ma], cfg[strategy][slow_ma])# ① 含流动性过滤 print(\n *60)print( ① 趋势策略 流动性过滤)print(*60)lf LiquidityFilter(lookback_dayscfg[liquidity][lookback_days],min_avg_amountcfg[liquidity][min_avg_amount],max_cvcfg[liquidity][max_cv],emergency_amountcfg[liquidity][emergency_amount])strat1 TrendLiquidityStrategy(liquidity_filterlf,fast_macfg[strategy][fast_ma],slow_macfg[strategy][slow_ma],max_positionscfg[strategy][max_positions],initial_capitalcfg[strategy][initial_capital])r1 run_backtest(strat1, close, amount_mtx, signals,cfg[backtest][start_date], cfg[backtest][end_date],enable_liquidityTrue)strat1.print_stats()# ② 无流动性过滤对比组print(\n *60)print( ② 趋势策略无流动性过滤)print(*60)lf_none LiquidityFilter(min_avg_amount0, max_cv999, emergency_amount0)strat2 TrendLiquidityStrategy(liquidity_filterlf_none,fast_macfg[strategy][fast_ma],slow_macfg[strategy][slow_ma],max_positionscfg[strategy][max_positions],initial_capitalcfg[strategy][initial_capital])r2 run_backtest(strat2, close, amount_mtx, signals,cfg[backtest][start_date], cfg[backtest][end_date],enable_liquidityFalse)strat2.print_stats()# 对比m1 calc_metrics(r1[nav])m2 calc_metrics(r2[nav])print_comparison(m1, m2)print(✅ 回测完成)if __name__ __main__:main()六、README.md 与使用说明# 通用趋势策略 成交额流动性过滤## 核心功能在通用趋势策略双均线基础上增加三层成交额流动性过滤1. 日均成交额 ≥ N 元2. 成交额变异系数 ≤ 阈值3. 单日成交额熔断保护## 安装bashpip install -r requirements.txt## 快速开始### 1. 准备数据**行情数据** data/daily_prices.csvcsvdate,code,open,high,low,close2022-01-03,000001,12.3,12.5,12.1,12.4...**成交额数据** data/daily_amounts.csvcsvdate,code,amount2022-01-03,000001,152345678...### 2. 配置流动性参数yamlliquidity:lookback_days: 20min_avg_amount: 20000000 # 2000 万/日max_cv: 0.8emergency_amount: 3000000 # 300 万熔断### 3. 运行bashpython main.py输出- 有/无流动性过滤的策略对比- 被拦截个股统计- 核心评价指标差异## 参数参考| 参数 | 宽松 | 中性 | 严格 ||------|------|------|------|| min_avg_amount | 500 万 | 2000 万 | 5000 万 || max_cv | 1.0 | 0.8 | 0.5 || emergency_amount | 100 万 | 300 万 | 500 万 |七、核心知识点卡片┌──────────────────────────────────────────────────────────────┐│ 趋势策略 流动性过滤 — 核心知识 │├────────────────┬─────────────────────────────────────────────┤│ 趋势策略 │ 双均线/突破/动量最通用的量化骨架 ││ 成交额过滤 │ 比市值更真实反映能不能卖出 ││ 三层结构 │ 日均门槛 → 稳定性 → 熔断保护 ││ 变异系数 │ CV std/mean衡量流动性稳定性 ││ 回测陷阱 │ 无流动性过滤的回测 ≈ 假设 ││ 正确集成顺序 │ 趋势信号 → 流动性过滤 → 下单 ││ 核心原则 │ 不能退出的策略不是策略是假设 │└────────────────┴─────────────────────────────────────────────┘八、免责声明与风险提示⚠️ 免责声明本代码仅供学习、研究与量化教学用途不构成任何投资建议或投资决策依据。模拟数据为随机数生成不代表任何真实标的历史或未来表现。⚠️ 风险提示- 流动性过滤会减少可选标的可能降低策略容量- 成交额数据本身可能被短期操纵对敲放量- 极端行情下流动性可能在持仓期间恶化静态过滤无法应对- 回测中信号为简化逻辑实盘需更完善的趋势判定- 历史流动性不保证未来流动性本文代码仅供学习与技术交流不构成任何投资建议股市有风险入市需谨慎利用AI解决实际问题如果你觉得这个工具好用欢迎关注长安牧笛