/~learn/s/d7c3e2f9
> 最后更新:2026-07-01 > > 本文档是选股子系统的总体设计。它直接复用现有截面选基的 7 层架构(Factor → Selector → Engine → 回测 → 推送),只是把 ETF 因子换成股票因子、ETF 池换成全 A 股池。
---
| 维度 | 决定 |
| ------ | ------ |
| 市场范围 | 全 A 股(5000+ 只),动态过滤 ST/停牌/次新股 |
| 频率 | 双周频(回测可验证优化,未来可切月频/周频对比) |
| 策略哲学 | 多因子融合(动量 + 稳定性 + 反追高 + 量价确认 + 流动性) |
| 仓位 | 10 只,核心持仓(5 只 60% 仓位)+ 卫星持仓(5 只 40% 仓位) |
| V1 因子 | 全部量价因子,无需财务数据,adata.stock.market.get_market() 即可 |
| V2 规划 | 加入财务因子(PE/PB/ROE/负债率),需接入 adata.stock.finance 模块 |
---
selection/
├── stock/ # 🆕 选股子系统(与 selection/ 平级)
│ ├── universe/
│ │ └── stock_pool.py # 全A股动态池(ST/停牌/次新过滤)
│ ├── factor/
│ │ ├── base.py # 复用 Factor 抽象基类
│ │ ├── momentum.py # 动量因子(多窗口收益 + R²趋势质量 + 回撤惩罚)
│ │ ├── stability.py # 波动率稳定性因子
│ │ ├── anti_overbought.py # 反追高因子(距均线偏离度惩罚)
│ │ ├── volume_confirm.py # 量价确认因子
│ │ ├── liquidity.py # 流动性因子(成交额 + 换手率)
│ │ └── diversification.py # 行业+市值分散化过滤
│ ├── selector/
│ │ ├── base.py # StockSelector 抽象基类
│ │ └── multi_factor.py # 多因子加权选择器
│ ├── engine.py # StockSelectionEngine 核心引擎
│ └── run_selection.py # 命令行入口
关键差异点(vs ETF 选基):
| 维度 | ETF 选基 | 股票选股 |
| ------ | --------- | --------- |
| 池子大小 | 107 只 | 5000+ 只(需实时过滤) |
| 数据源 | 东方财富基金 API / Tushare | 东方财富股票 API(复用 adata.stock.market) |
| 因子 | 5 个(动量/夏普/反追高/稳定性/估值) | V1: 5 个纯量价因子 |
| 分散化 | 同指数去重+相关性+大类限数 | 行业去重+相关性+市值分层 |
| 基准 | 沪深 300 | 中证全指 |
| Market Regime | 用 510300.SH 检测 | 可用沪深 300 指数判断 |
---
全 A 股实时池 = 东方财富全市场股票列表 − 过滤条件
| 过滤条件 | 原因 |
| ---------- | ------ |
| ST / \*ST 股票 | 退市风险,流动性差 |
| 上市不满 60 个交易日 | 新股波动剧烈,没有足够历史数据 |
| 当日停牌 | 无法交易 |
| 日均成交额 < 2000 万(近 20 日) | 流动性不足,冲击成本大 |
| 股价 < 2 元 | 低价股风险 + 涨停板制度扭曲 |
使用东方财富行业分类(申万一级),用于分散化过滤。代码参考:adata.stock.info 模块。
---
所有 V1 因子都是纯量价因子,只需 OHLCV 数据,不依赖财务报表。
问的问题:「过去一段时间涨了多少?涨得干净吗?」
raw_momentum = 40日收益 × 0.5 + 80日收益 × 0.5
final_momentum = raw × max(R²趋势质量, 0.3) × 回撤惩罚
问的问题:「波动大不大?涨得稳不稳?」
stability = 1 - (近40日波动率 / 近120日波动率)
问的问题:「是不是涨太疯了该回避?」
偏离度 = (当前价 - 60日均价) / 60日均价
penalty = 1 - min(max(偏离度 - 0.15, 0) / 0.35, 1)
问的问题:「上涨有没有成交量配合?」
量比 = 近20日均量 / 近60日均量
价的方向 = sign(近20日收益)
量价确认 = sign(价) × (量比 - 1)
问的问题:「交易够不够活跃?」
流动性分 = rank(近20日日均成交额) × 0.5 + rank(近20日日均换手率) × 0.5
V1 先用等权(20% × 5),后续迭代时再引入 Adaptive Weight Engine。
---
复用 MultiFactorETFSelector 的逻辑:
factor.normalize(series, method='rank') → 百分位排名(0-1 区间)class MultiFactorStockSelector(StockSelector):
def __init__(self, factors: List[Factor], weights: List[float]):
self.factors = factors
self.weights = weights
def select(self, data_dict, date, top_n=10, regime_info=None):
scores = pd.DataFrame(index=list(data_dict.keys()))
for factor, weight in zip(self.factors, self.weights):
raw = factor.calculate(data_dict, date)
normalized = factor.normalize(raw, method='rank')
if normalized.std() > 0:
scores[factor.name] = normalized * weight
ranked = scores.sum(axis=1).sort_values(ascending=False)
return ranked.head(top_n).index.tolist(), ranked
---
同一申万一级行业最多 2 只,防止单一行业扎堆。
行业上限 = {
'电子': 2, '医药生物': 2, '电力设备': 2, '计算机': 2,
... 其他行业默认 2
}
贪心算法:从高分到低分遍历,新候选股与已选股 40 日收益率相关性 > 0.7 则跳过。
按总市值分三层,每层至少 2 只:
---
StockSelectionEngine 对标 ETFSelectionEngine:
class StockSelectionEngine:
def __init__(self, stock_pool, selector, data_fetcher):
...
def fetch_all_data(self, years_back=2):
"""从东方财富拉取全 A 股日 K,过滤 ST/停牌/次新/低流动性"""
...
def build_snapshot(self, date, lookback_days=250):
"""截取每只股票近 N 日 OHLCV 数据"""
...
def run_selection(self, date, top_n=10):
"""过滤股票池 → 打分 → 分散化 → 输出 Top N"""
...
股票用 adata.stock.market.get_market() 拉取,代码格式为 000001.SZ、600036.SH。
全 A 股 5000+ 只,拉取压力大。V1 策略:
adata.stock.info)---
V1 先不做完整回测框架,用 CrossSectionalBacktester 的逻辑适配:
---
cross_section_weekly_snapshot 表或新建 stock 表)---
以下选股策略方向值得后续研究,当前文档只记录方向,每个方向需要独立设计文档和回测验证:
adata.stock.finance,用 PE/PB/PS/股息率等估值因子选股