01-研究工程与治理
预计学习时间：4-6 小时
难度：⭐⭐⭐⭐⭐
核心问题：你发现了一个”能赚钱”的策略——但你确定它真的能赚钱吗？
从一个真实场景出发
你是一个量化研究员，花了两周时间构建了一个新的动量因子。回测结果令人兴奋：
- 年化收益 35%
- Sharpe 2.8
- 最大回撤 8%
- 回撤恢复期从未超过 3 个月
你兴冲冲地写了一份报告，提交给投资委员会。但 CRO（首席风险官）只问了三个问题：
- 你用的数据包含退市股票吗？
- 你的交易成本模型是怎么设的？
- 这个因子的月均换手率是多少？容量是多少？
你突然意识到：你从来没有考虑过这三个问题。
更残酷的是，当你回去仔细检查之后发现：
- 你的数据只包含当前存活的股票 → 幸存者偏差：把退市的垃圾股排除了，当然”低价值股”表现好
- 你的回测用的是 5bp 的交易成本 → 低估成本：实际上这个因子月换手率 200%，以真实滑点计算，收益几乎为零
- 你的因子容量只有 5000 万 → 无法规模化：即使真的有效，对基金的整体贡献也微乎其微
这不是虚构的故事，而是几乎每个量化研究员都经历过的”假 Alpha”教训。
这章讲的就是：如何通过系统化的治理流程，防止这种”假 Alpha”进入实盘。
一、命题治理（Idea Governance）
1.1 核心思想：从”灵感”到”命题”需要形式化
个人研究者最常见的习惯是：想到一个想法 → 直接写代码回测 → 看结果决定要不要继续。
机构研究不是这么做的。在机构里，一个想法要变成”可研究的命题”，需要经过形式化。
想法 → 命题的转化过程：
灵感阶段 命题阶段
┌────────────────────┐ ┌────────────────────────────┐
│ "我觉得低波动股票 │ │ 命题：在控制市值和行业暴露后，│
│ 长期表现更好" │ ──→ │ 低波动率股票的超额收益是否 │
│ │ │ 显著为正？ │
│ 模糊的直觉 │ │ │
│ 不可验证 │ │ 明确的假设 │
│ 没有边界条件 │ │ 可检验的表述 │
└────────────────────┘ │ 有明确的适用范围 │
└────────────────────────────┘
一个”好命题”需要回答四个问题：
| 问题 | 好的答案 | 不好的答案 |
|---|---|---|
| 为什么有效？ | 低波动率股票因为杠杆约束和机构基准竞争被系统性低估 | ”回测数据就是这样” |
| 在什么市场有效？ | 成熟市场，流动性足够的中大盘股 | ”所有股票都行” |
| 可能失效的条件？ | 当大量资金涌入低波动策略导致拥挤时 | ”不会失效” |
| 容量大约多大？ | 根据市场流动性和持仓集中度估算 | 没想过 |
1.2 命题评审标准
机构对命题有明确的评审标准，通常从三个维度打分：
维度一：经济学逻辑（权重 40%）
| 评分 | 标准 |
|---|---|
| A | 有清晰的理论基础（风险补偿、行为偏差、市场摩擦），有学术文献支撑 |
| B | 有合理的经济学故事，但理论支撑不够强 |
| C | 纯数据驱动，没有经济学解释 |
| D | 甚至违反基本的金融学原理（比如”永远上涨”的策略） |
维度二：统计可行性（权重 30%）
| 评分 | 标准 |
|---|---|
| A | 数据容易获取、质量高、样本量充足、不需要复杂的数据处理 |
| B | 数据可以获取，但需要一定的清洗和处理 |
| C | 数据获取困难，或质量存疑，或样本量不足 |
| D | 数据几乎不可能获取，或质量无法保证 |
维度三：容量评估（权重 30%）
| 评分 | 标准 |
|---|---|
| A | 预估容量 > 5 亿，对组合有实质性贡献 |
| B | 预估容量 5000 万 - 5 亿，可以作为辅助策略 |
| C | 预估容量 < 5000 万，对组合贡献有限 |
| D | 容量极小，不值得投入研发资源 |
1.3 命题生命周期管理
命题不是一锤子买卖。它有完整的生命周期：
命题生命周期：
提出 ──→ 验证 ──→ 上线 ──→ 监控 ──→ 退市
│ │ │ │ │
↓ ↓ ↓ ↓ ↓
评审 样本内 Gate 持续 触发退
打分 样本外 检查 跟踪 市标准
验证 指标
┌─────────────────────────────────────────────┐
│ 状态机： │
│ │
│ DRAFT ──→ REVIEW ──→ VALIDATING ──→ LIVE │
│ ↑ │ │
│ │ ↓ │
│ └──── REJECTED RETIRED
│ │ │
│ └──→ REVISION ──→ REVIEW │
└─────────────────────────────────────────────┘
每个阶段的具体要求：
| 阶段 | 产出物 | 通过条件 |
|---|---|---|
| DRAFT | 命题文档（动机、经济学逻辑、数据需求） | 评审分数 >= B- |
| REVIEW | 同行评审意见 | 至少一位高级研究员同意进入验证 |
| VALIDATING | 样本内结果 + 样本外结果 + 稳健性检验 | 见 5.1 Gate 检查点 |
| LIVE | 上线后的监控报告 | 持续满足运行指标 |
| RETIRED | 退市分析报告 | 触发退市条件（见 5.3） |
1.4 命题优先级排序
资源有限，不可能同时研究所有命题。机构通常用”影响力-可行性”矩阵来排序：
命题优先级矩阵：
可行性（高）
│
┌───────────┼───────────┐
│ Quick │ │
│ Wins │ Major │
│ │ Projects │
│ 立即执行 │ │
影响 ├───────────┼───────────┤ 影响
（高）│ │ │ （低）
│ Fill-ins │ Time │
│ │ Sinks │
│ 有空再做 │ │
│ │ 不做 │
└───────────┼───────────┘
│
可行性（低）
| 象限 | 决策 | 典型例子 |
|---|---|---|
| Quick Wins | 立即执行 | 有经济学逻辑、数据现成的小改进 |
| Major Projects | 重点投入 | 新策略类型、需要新数据源的研究 |
| Fill-ins | 有空再做 | 微小的参数优化、非核心的探索 |
| Time Sinks | 不做 | 数据不可获取、经济学逻辑薄弱的”猎奇”想法 |
1.5 Python：命题管理系统
import numpy as np
import pandas as pd
from enum import Enum
from dataclasses import dataclass, field
from datetime import datetime
from typing import Optional
# ============================================================
# 命题治理系统模拟
# ============================================================
class PropositionStatus(Enum):
DRAFT = "draft"
REVIEW = "review"
VALIDATING = "validating"
LIVE = "live"
RETIRED = "retired"
REJECTED = "rejected"
REVISION = "revision"
@dataclass
class Proposition:
"""研究命题"""
id: str
title: str
status: PropositionStatus = PropositionStatus.DRAFT
# 评审打分（1-5）
economics_score: float = 0.0 # 经济学逻辑
statistics_score: float = 0.0 # 统计可行性
capacity_score: float = 0.0 # 容量评估
# 元信息
author: str = ""
created_at: datetime = field(default_factory=datetime.now)
reviewer: Optional[str] = None
# 验证结果
in_sample_sharpe: Optional[float] = None
out_sample_sharpe: Optional[float] = None
# 实盘结果
live_sharpe: Optional[float] = None
live_turnover: Optional[float] = None
def composite_score(self) -> float:
"""综合评分：经济学 40% + 统计 30% + 容量 30%"""
return (0.4 * self.economics_score +
0.3 * self.statistics_score +
0.3 * self.capacity_score)
def can_advance(self) -> tuple:
"""判断命题是否可以推进到下一阶段"""
if self.status == PropositionStatus.DRAFT:
if self.composite_score() >= 3.0:
return True, "综合评分达标，可进入评审"
return False, f"综合评分 {self.composite_score():.1f} < 3.0"
if self.status == PropositionStatus.REVIEW:
if self.reviewer is not None:
return True, "已有评审人，可进入验证"
return False, "需要指定评审人"
if self.status == PropositionStatus.VALIDATING:
if (self.in_sample_sharpe and self.out_sample_sharpe and
self.out_sample_sharpe > 0.5):
return True, "样本外 Sharpe 达标"
return False, "样本外验证未通过"
return False, "当前状态无法推进"
class PropositionGovernance:
"""命题治理系统"""
def __init__(self):
self.propositions: dict[str, Proposition] = {}
def submit(self, prop: Proposition) -> dict:
"""提交新命题"""
can_advance, reason = prop.can_advance()
if not can_advance:
return {"status": "rejected", "reason": reason}
self.propositions[prop.id] = prop
prop.status = PropositionStatus.REVIEW
return {"status": "accepted", "reason": reason}
def prioritize(self) -> pd.DataFrame:
"""按优先级排序所有活跃命题"""
active = [p for p in self.propositions.values()
if p.status in (PropositionStatus.DRAFT,
PropositionStatus.REVIEW,
PropositionStatus.VALIDATING)]
if not active:
return pd.DataFrame()
data = [{
'ID': p.id,
'标题': p.title,
'综合评分': p.composite_score(),
'经济学': p.economics_score,
'统计可行性': p.statistics_score,
'容量': p.capacity_score,
'状态': p.status.value
} for p in active]
df = pd.DataFrame(data)
return df.sort_values('综合评分', ascending=False)
# ============================================================
# 使用示例
# ============================================================
governance = PropositionGovernance()
# 提交三个命题
props = [
Proposition(
id="PROP-001",
title="基于异质波动的选股因子",
economics_score=4.0, # 有理论支撑
statistics_score=3.5, # 数据现成
capacity_score=4.0, # 容量充足
author="张三"
),
Proposition(
id="PROP-002",
title="基于社交媒体情绪的日内预测",
economics_score=2.0, # 逻辑薄弱
statistics_score=2.5, # 数据获取难
capacity_score=1.5, # 容量极小
author="李四"
),
Proposition(
id="PROP-003",
title="多因子动态权重优化",
economics_score=4.5, # 理论扎实
statistics_score=4.0, # 数据充足
capacity_score=3.0, # 中等容量
author="王五"
),
]
for prop in props:
result = governance.submit(prop)
print(f"{'='*50}")
print(f"命题: {prop.title}")
print(f" 综合评分: {prop.composite_score():.1f}")
print(f" 结果: {result['status']} - {result['reason']}")
print(f"\n{'='*50}")
print("命题优先级排序:")
print(governance.prioritize().to_string(index=False))二、数据治理（Data Governance）
2.1 核心思想：没有好数据，一切都是空谈
量化研究有一个铁律：Garbage in, Garbage out。
无论你的模型多精巧，如果输入的数据有问题，输出就一定有问题。而数据的”问题”往往不是明显的错误（比如价格是负数），而是微妙的偏差——前视偏差、幸存者偏差、数据清洗引入的偏差。
这些偏差有多隐蔽？来看一个真实的例子。
案例：一个”永远不会亏”的策略
某研究员发现了一个规律：每天买入前一天的涨幅最大的 10 只股票，年化收益高达 50%。但这个策略有一个致命问题——他在计算”前一天涨幅”时，用的是当天收盘后才能获得的复权价格。也就是说，他用今天的信息做了昨天的决策。
这就是前视偏差（Look-ahead Bias）。
前视偏差示意图：
正确的时序： 有偏差的时序：
───────── ─────────
T 日收盘 T 日收盘
↓ ↓
计算 T 日的信号 用 T 日的复权因子
↓ ↓
T+1 日开盘买入 计算 T 日的信号 ← 错！
↓
T+1 日开盘买入
正确：用 T 日及之前的信息决定 T+1 的交易
错误：用 T 日的信息"回填"到 T 日的信号计算中
2.2 数据源管理与版本控制
机构对数据的管理远比个人研究者严格：
| 维度 | 个人研究者 | 机构做法 |
|---|---|---|
| 数据存储 | 本地 CSV 文件 | 数据仓库 + 版本控制 |
| 数据更新 | 手动下载 | 自动化管道 + 变更日志 |
| 数据溯源 | ”从某网站下的” | 数据血缘：每条数据可追溯到源 |
| 版本管理 | 覆盖更新 | DVC / 数据快照 / 变更通知 |
| 访问控制 | 无 | 研究环境隔离，生产数据只读 |
数据版本控制的核心原则：
数据版本控制的三条铁律：
1. 永远不要修改原始数据
原始数据是"真值"，发现错误也不改原始数据，
而是在下游增加"数据修正层"并记录原因
2. 每次数据处理都要记录完整的环境
包括：代码版本、依赖版本、参数配置、数据版本
目的：任何人拿到这些信息都能复现结果
3. 数据变更必须通知所有使用者
数据供应商修正历史数据时（非常常见），
必须通知所有依赖该数据的研究员
2.3 数据质量校验
数据质量校验从四个维度进行：
完整性（Completeness）
检查项目：
- 缺失值比例（超过 5% 需要标记）
- 缺失模式（随机缺失 vs 系统性缺失）
- 时间覆盖（是否存在"数据断档"）
- 股票覆盖（是否所有股票在所有时间都有数据）
一致性（Consistency）
检查项目：
- 同一只股票在不同数据源中的差异
- 同一天不同频率数据（日频 vs 分钟频）的对齐
- 复权因子的一致性（前复权 vs 后复权 vs 不复权）
- 行业分类的一致性（申万 vs 中信 vs GICS）
准确性（Accuracy）
检查项目：
- 异常值检测（价格 = 0？成交量 = 0？收益率 > 100%？）
- 交叉验证（两个数据源的同一指标是否一致）
- 已知错误列表检查（供应商公布的数据修正）
- 逻辑校验（开盘价不可能远高于收盘价 + 涨跌停幅度）
时效性（Timeliness）
检查项目：
- 数据延迟（日频数据通常 T+1 可用，有没有 T+0？）
- 数据修正（财务数据是否会被修正？修正频率？）
- 点数据 vs 区间数据（实时数据 vs 收盘后数据）
2.4 系统性偏差检查
这是数据治理最重要的部分。三种最常见的偏差：
幸存者偏差（Survivorship Bias）
import numpy as np
import pandas as pd
np.random.seed(42)
# ============================================================
# 模拟幸存者偏差的影响
# ============================================================
n_stocks = 500
n_months = 60 # 5 年
# 初始市值
initial_market_cap = np.random.lognormal(16, 1, n_stocks)
# 每月收益：大公司和小公司有不同的分布
monthly_returns = np.zeros((n_months, n_stocks))
for t in range(n_months):
for s in range(n_stocks):
# 小公司：收益波动更大，有更多极端亏损
if initial_market_cap[s] < np.median(initial_market_cap):
monthly_returns[t, s] = np.random.normal(-0.002, 0.08)
else:
monthly_returns[t, s] = np.random.normal(0.002, 0.04)
# 模拟退市：连续亏损超过 50% 的股票在当年退市
alive = np.ones(n_stocks, dtype=bool)
for t in range(n_months):
for s in range(n_stocks):
if alive[s]:
cumulative = np.prod(1 + monthly_returns[:t+1, s]) - 1
if cumulative < -0.5:
alive[s] = False
monthly_returns[t:, s] = np.nan # 退市后没有数据
# ============================================================
# 对比：有偏差 vs 无偏差的结果
# ============================================================
# 有幸存者偏差的数据：只包含当前存活的公司
survivor_returns = monthly_returns[:, alive]
# 无偏差的数据：包含所有公司（退市的也包含）
full_returns = monthly_returns.copy()
full_returns[np.isnan(full_returns)] = -0.1 # 退市当月跌 10%
# 按初始市值分组
small_mask = initial_market_cap < np.median(initial_market_cap)
large_mask = ~small_mask
def calc_annual_return(returns):
"""计算年化收益"""
valid = returns[~np.isnan(returns)]
if len(valid) == 0:
return np.nan
return (np.prod(1 + valid) ** (12 / len(valid)) - 1)
print("=" * 60)
print("幸存者偏差的影响")
print("=" * 60)
# 小市值组
small_survivor = calc_annual_return(survivor_returns[:, small_mask])
small_full = calc_annual_return(full_returns[:, small_mask])
# 大市值组
large_survivor = calc_annual_return(survivor_returns[:, large_mask])
large_full = calc_annual_return(full_returns[:, large_mask])
print(f"\n{'指标':<20} {'有幸存者偏差':>15} {'无偏差':>15}")
print("-" * 50)
print(f"{'小市值年化收益':<18} {small_survivor:>14.1%} {small_full:>14.1%}")
print(f"{'大市值年化收益':<16} {large_survivor:>14.1%} {large_full:>14.1%}")
print(f"{'小市值溢价':<20} {small_survivor - large_survivor:>14.1%} "
f"{small_full - large_full:>14.1%}")
print(f"\n结论：幸存者偏差让小市值溢价看起来比实际大")
print(f"退市公司通常是表现差的（小市值更多），排除它们会高估收益")前视偏差（Look-ahead Bias）
import numpy as np
import pandas as pd
np.random.seed(42)
# ============================================================
# 模拟前视偏差的影响
# ============================================================
n_stocks = 300
n_days = 252 # 一年
# 模拟日收益率
returns = np.random.normal(0.0005, 0.02, (n_days, n_stocks))
# 模拟复权因子：每天收盘后更新
# 复权因子反映了当天的分红和拆股
adjust_factors = np.ones((n_days, n_stocks))
for t in range(1, n_days):
adjust_factors[t] = adjust_factors[t-1] * np.random.choice(
[1.0, 0.98, 0.99], n_stocks, p=[0.9, 0.05, 0.05]
)
# ============================================================
# 正确 vs 错误的计算方式
# ============================================================
# 错误：用 T 日的复权因子计算 T 日的信号
# （复权因子是 T 日收盘后才更新的）
wrong_signal = np.zeros((n_days, n_stocks))
for t in range(n_days):
# 用 T 日的复权因子调整 T 日之前的价格
wrong_signal[t] = adjust_factors[t] * np.random.normal(0, 0.01, n_stocks)
# 正确：用 T-1 日的复权因子计算 T 日的信号
correct_signal = np.zeros((n_days, n_stocks))
for t in range(1, n_days):
correct_signal[t] = adjust_factors[t-1] * np.random.normal(0, 0.01, n_stocks)
# 用信号预测次日的收益
wrong_ic = np.array([
np.corrcoef(wrong_signal[t], returns[min(t+1, n_days-1)])[0, 1]
for t in range(n_days - 1)
])
correct_ic = np.array([
np.corrcoef(correct_signal[t], returns[min(t+1, n_days-1)])[0, 1]
for t in range(1, n_days - 1)
])
print("=" * 50)
print("前视偏差对 IC 的影响")
print("=" * 50)
print(f"正确方式的平均 IC: {np.nanmean(correct_ic):.4f}")
print(f"错误方式的平均 IC: {np.nanmean(wrong_ic):.4f}")
print(f"IC 膨胀: {np.nanmean(wrong_ic) - np.nanmean(correct_ic):.4f}")
print(f"\n即使数据本身没有问题，时序对齐错误也会产生虚假信号")数据清洗偏差（Data-snooping via Cleaning）
数据清洗偏差的经典场景：
1. 研究员发现某个因子在 2018-2022 年有效
2. 仔细检查后，发现 2019 年有一段数据异常
3. "合理地"剔除了那段异常数据
4. 因子在剩余时间段表现更好了
5. 但这个"更好"是真实的还是因为你剔除了不利的数据？
关键问题：你剔除数据的标准是什么？
- 如果标准是"让因子看起来更好" → 数据清洗偏差
- 如果标准是独立于因子结果的（比如数据供应商确认错误） → 合理的
2.5 Python：数据质量检查框架
import numpy as np
import pandas as pd
from dataclasses import dataclass
from typing import Optional
# ============================================================
# 数据质量检查框架
# ============================================================
@dataclass
class QualityCheckResult:
"""单项检查结果"""
check_name: str
passed: bool
severity: str # "error", "warning", "info"
details: str = ""
affected_rows: int = 0
class DataQualityChecker:
"""数据质量检查器"""
def __init__(self, df: pd.DataFrame):
self.df = df
self.results: list[QualityCheckResult] = []
def check_completeness(self, threshold: float = 0.05) -> QualityCheckResult:
"""检查缺失值"""
total = self.df.shape[0] * self.df.shape[1]
missing = self.df.isnull().sum().sum()
missing_ratio = missing / total
passed = missing_ratio < threshold
severity = "error" if missing_ratio > 0.1 else "warning" if missing_ratio > threshold else "info"
details = f"缺失值比例: {missing_ratio:.2%}"
return QualityCheckResult(
check_name="完整性",
passed=passed,
severity=severity,
details=details,
affected_rows=int(missing)
)
def check_price_validity(self, price_col: str) -> QualityCheckResult:
"""检查价格合理性"""
issues = 0
issues_details = []
# 价格 <= 0
neg_count = (self.df[price_col] <= 0).sum()
if neg_count > 0:
issues += neg_count
issues_details.append(f"非正价格: {neg_count} 条")
# 日收益率 > 30%（排除涨跌停）
returns = self.df[price_col].pct_change()
extreme = (returns.abs() > 0.30).sum()
if extreme > 0:
issues += extreme
issues_details.append(f"极端收益(>30%): {extreme} 条")
return QualityCheckResult(
check_name="价格合理性",
passed=issues == 0,
severity="error" if issues > 10 else "warning",
details="; ".join(issues_details) if issues_details else "通过",
affected_rows=issues
)
def check_lookahead(self, signal_col: str, return_col: str) -> QualityCheckResult:
"""检查是否存在前视偏差（信号与同期收益的相关性）"""
corr = self.df[signal_col].corr(self.df[return_col])
suspicious = abs(corr) > 0.1
return QualityCheckResult(
check_name="前视偏差检查",
passed=not suspicious,
severity="error" if suspicious else "info",
details=f"信号与同期收益相关系数: {corr:.4f}",
affected_rows=0
)
def check_time_gaps(self, date_col: str, freq: str = "D") -> QualityCheckResult:
"""检查时间连续性"""
dates = pd.to_datetime(self.df[date_col])
expected = pd.date_range(dates.min(), dates.max(), freq=freq)
missing_dates = set(expected) - set(dates.unique())
gap_count = len(missing_dates)
return QualityCheckResult(
check_name="时间连续性",
passed=gap_count == 0,
severity="warning" if gap_count > 5 else "info",
details=f"缺失交易日: {gap_count} 天",
affected_rows=gap_count
)
def run_all_checks(self, price_col: str = "close",
signal_col: Optional[str] = None,
return_col: Optional[str] = None,
date_col: str = "date") -> pd.DataFrame:
"""运行所有检查并生成报告"""
self.results = [
self.check_completeness(),
self.check_price_validity(price_col),
self.check_time_gaps(date_col),
]
if signal_col and return_col:
self.results.append(self.check_lookahead(signal_col, return_col))
report = pd.DataFrame([{
'检查项': r.check_name,
'通过': 'PASS' if r.passed else 'FAIL',
'严重度': r.severity,
'详情': r.details,
'影响行数': r.affected_rows
} for r in self.results])
return report
# ============================================================
# 使用示例
# ============================================================
np.random.seed(42)
# 模拟有问题的数据
n_rows = 500
dates = pd.date_range("2022-01-01", periods=n_rows, freq="D")
# 故意制造一些问题
prices = np.random.lognormal(4, 0.3, n_rows).astype(float)
prices[10] = -1.0 # 非法价格
prices[50] = 0.0 # 非法价格
prices[200] *= 5 # 极端值
signal = np.random.normal(0, 1, n_rows)
# 故意让信号和同期收益有微弱相关性（模拟前视偏差）
returns = 0.05 * signal + np.random.normal(0, 0.02, n_rows)
df = pd.DataFrame({
'date': dates,
'close': prices,
'signal': signal,
'return': returns
})
# 运行检查
checker = DataQualityChecker(df)
report = checker.run_all_checks(
price_col='close',
signal_col='signal',
return_col='return',
date_col='date'
)
print("=" * 60)
print("数据质量检查报告")
print("=" * 60)
print(report.to_string(index=False))三、特征治理（Feature Governance）
3.1 核心思想：特征是研究的基本单元，必须可追踪、可复现
在量化研究中，“特征”（Feature）是你用来预测收益的变量。一个因子（如动量、价值）通常由一个或多个特征组成。
个人研究者构建特征的习惯通常是：写一段代码 → 算出数值 → 直接用。问题在于：
- 三个月后你自己都不记得这段代码是什么逻辑了
- 换台电脑，代码跑不出来了（环境不一致）
- 另一个研究员想验证你的结果，完全无法复现
特征治理要解决的就是这些问题。
3.2 特征命名规范与文档化
好的特征命名规范能让任何人一眼看出这个特征是什么：
命名规范：{市场}_{频率}_{类别}_{具体定义}_{版本}
例子：
- CN_D_FACTOR_MOMENTUM_20D_V1 (A股 日频 因子 动量 20日 版本1)
- US_M_FACTOR_VALUE_PB_V2 (美股 月频 因子 价值 市净率 版本2)
- CN_D_RISK_VOLATILITY_60D_V1 (A股 日频 风险 波动率 60日 版本1)
反面例子：
- my_factor (你的什么因子？)
- test_v3 (测试什么？第几个版本？)
- factor_new (新的因子？多新？)
每个特征必须有文档记录：
| 字段 | 内容 |
|---|---|
| 名称 | 遵循命名规范的唯一标识 |
| 定义 | 精确的数学定义 |
| 经济学含义 | 为什么这个特征可能预测收益 |
| 输入数据 | 依赖哪些原始数据字段 |
| 计算逻辑 | 伪代码或核心代码片段 |
| 作者 | 创建者 |
| 创建时间 | 首次创建的日期 |
| 版本历史 | 每次修改的变更说明 |
| 预期性质 | IC 方向、衰减速度、与已知因子的相关性 |
3.3 特征上线流程
特征上线流程：
开发 → 自测 → 测试环境 → 同行评审 → 审批 → 生产环境
│ │ │ │ │ │
↓ ↓ ↓ ↓ ↓ ↓
研究 单元 集成 另一位 治理 自动化
人员 测试 测试 研究员 委员会 管道
检查 独立验证 审批
每个阶段的检查点：
┌─────────────────────────────────────────────────┐
│ 自测： │
│ - 代码能正确运行？ │
│ - 边界情况处理了吗（停牌、涨跌停、新股）？ │
│ - 输出分布合理吗？ │
├─────────────────────────────────────────────────┤
│ 测试环境： │
│ - 与已有特征的兼容性？ │
│ - 计算耗时可接受吗？ │
│ - 内存使用合理吗？ │
├─────────────────────────────────────────────────┤
│ 同行评审： │
│ - 另一位研究员能独立复现结果？ │
│ - 经济学逻辑站得住脚？ │
│ - IC/ICIR 符合预期？ │
├─────────────────────────────────────────────────┤
│ 审批： │
│ - 特征文档完整？ │
│ - 命名规范符合？ │
│ - 上线风险评估？ │
└─────────────────────────────────────────────────┘
3.4 特征监控
特征上线只是开始。上线后需要持续监控：
| 监控指标 | 含义 | 告警阈值 |
|---|---|---|
| 分布漂移（Distribution Drift） | 特征的均值/方差是否发生系统性变化 | 均值偏移 > 2 个标准差 |
| IC 衰减（IC Decay） | 特征的预测力是否在下降 | 滚动 12 个月 ICIR < 0.5 |
| 缺失率 | 特征的计算是否有异常 | 缺失率突然 > 5% |
| 计算耗时 | 特征计算是否变慢 | 耗时 > 历史均值 + 2 个标准差 |
| 极端值比例 | 特征是否有异常波动 | 极端值比例 > 历史均值 + 3 个标准差 |
3.5 Python：特征监控与衰减检测
import numpy as np
import pandas as pd
np.random.seed(42)
# ============================================================
# 模拟特征衰减监控
# ============================================================
n_months = 120 # 10 年
n_stocks = 300
# 模拟 IC 序列
# 前 5 年：特征有效，IC 稳定
# 第 5-7 年：开始衰减
# 第 7-10 年：几乎失效
ic_series = np.zeros(n_months)
for t in range(n_months):
if t < 60:
# 有效阶段
ic_series[t] = np.random.normal(0.05, 0.03)
elif t < 84:
# 衰减阶段
decay = (t - 60) / 24
ic_series[t] = np.random.normal(0.05 * (1 - 0.6 * decay), 0.03)
else:
# 失效阶段
ic_series[t] = np.random.normal(0.01, 0.03)
dates = pd.date_range("2016-01-01", periods=n_months, freq="M")
ic_df = pd.DataFrame({'date': dates, 'IC': ic_series})
class FeatureMonitor:
"""特征监控器"""
def __init__(self, ic_series: pd.Series, name: str):
self.ic = ic_series
self.name = name
self.alerts: list[str] = []
def check_ic_decay(self, window: int = 12) -> dict:
"""检查 IC 衰减"""
rolling_icir = (self.ic.rolling(window).mean() /
self.ic.rolling(window).std())
current_icir = rolling_icir.iloc[-1]
# 分阶段分析
n = len(self.ic)
phase1_icir = (self.ic[:n//3].mean() / self.ic[:n//3].std())
phase2_icir = (self.ic[n//3:2*n//3].mean() / self.ic[n//3:2*n//3].std())
phase3_icir = (self.ic[2*n//3:].mean() / self.ic[2*n//3:].std())
result = {
'特征名称': self.name,
'当前滚动ICIR': f"{current_icir:.2f}",
'第一阶段ICIR': f"{phase1_icir:.2f}",
'第二阶段ICIR': f"{phase2_icir:.2f}",
'第三阶段ICIR': f"{phase3_icir:.2f}",
}
if current_icir < 0.5:
result['告警'] = 'ICIR 低于 0.5，特征可能失效'
elif current_icir < 1.0:
result['告警'] = 'ICIR 低于 1.0，需要关注'
else:
result['告警'] = '正常'
return result
def check_distribution_drift(self, current_values: np.ndarray,
reference_values: np.ndarray) -> dict:
"""检查分布漂移"""
current_mean = np.mean(current_values)
ref_mean = np.mean(reference_values)
ref_std = np.std(reference_values)
drift = abs(current_mean - ref_mean) / ref_std
return {
'特征名称': self.name,
'参考均值': f"{ref_mean:.4f}",
'当前均值': f"{current_mean:.4f}",
'漂移程度': f"{drift:.2f} 个标准差",
'状态': '严重漂移' if drift > 3 else
'轻微漂移' if drift > 2 else '正常'
}
# ============================================================
# 使用示例
# ============================================================
monitor = FeatureMonitor(ic_df['IC'], '动量_20D')
print("=" * 60)
print("特征衰减监控报告")
print("=" * 60)
result = monitor.check_ic_decay()
for k, v in result.items():
print(f" {k}: {v}")
# 分布漂移检查
current_feature_values = np.random.normal(0.02, 0.5, 1000)
reference_feature_values = np.random.normal(0.0, 0.5, 1000)
print(f"\n分布漂移检查:")
drift_result = monitor.check_distribution_drift(
current_feature_values, reference_feature_values
)
for k, v in drift_result.items():
print(f" {k}: {v}")3.6 特征依赖关系管理
特征之间往往存在依赖关系。管理不善会导致”级联失效”：
特征依赖关系示例：
原始数据层
├── 日频行情数据 (OHLCV)
├── 财务报表数据
└── 分析师预期数据
│
特征层
├── 波动率_60D ←── 依赖：日频行情
├── 资产负债率 ←── 依赖：财务报表
├── 分析师修正幅度 ←── 依赖：分析师预期
├── 低波动_Zscore ←── 依赖：波动率_60D + 行业市值中性化
└── 质量_综合评分 ←── 依赖：资产负债率 + ROE + 现金流
如果"日频行情数据"出问题：
→ 波动率_60D 失效
→ 低波动_Zscore 失效
→ 依赖低波动因子的策略全部受影响
依赖关系管理的原则：
1. 画出完整的特征依赖图
2. 上游数据变更必须评估下游影响
3. 关键上游数据有备份方案
四、回测治理（Backtest Governance）
4.1 核心思想：回测结果是”证据”，不是”证明”
这是量化研究最重要的认知转变之一：
回测结果好 ≠ 策略真的有效。回测结果好只能说明策略在过去的数据上表现好。
过拟合有多严重？Lopez de Prado 在《Advances in Financial Machine Learning》中提出了一个令人警醒的比喻：如果你在 1000 个随机策略中选最好的一个，它的回测表现一定会非常好——但它只是一个随机的策略。
回测过拟合的概率：
假设你有 N 个策略，每个策略的真实 Sharpe = 0（纯随机）
你选出回测 Sharpe 最高的那个策略
它的回测 Sharpe 期望值是多少？
E[Best Sharpe] ≈ sqrt(2 * ln(N) * T) / sqrt(T)
其中 T 是回测的月数
┌─────────────────────────────────────────┐
│ 试了 10 个策略，回测 5 年： │
│ 最高 Sharpe ≈ 0.9 │
│ 试了 100 个策略，回测 5 年： │
│ 最高 Sharpe ≈ 1.4 │
│ 试了 1000 个策略，回测 5 年： │
│ 最高 Sharpe ≈ 1.7 │
│ │
│ 但它们全是假的！ │
└─────────────────────────────────────────┘
4.2 回测标准化框架
为了防止”每个研究员用自己的回测框架”导致的不可比、不可复现问题，机构通常建立统一的回测框架：
回测标准化框架的三个统一：
┌────────────────────────────────────────────────────────┐
│ 统一回测框架 │
├────────────────────────────────────────────────────────┤
│ │
│ 1. 统一成本模型 │
│ ├── 佣金：按实际费率 │
│ ├── 印花税：按实际税率 │
│ ├── 滑点：按交易金额的函数（不是固定值） │
│ └── 冲击成本：按持仓比例和市场流动性的函数 │
│ │
│ 2. 统一数据源 │
│ ├── 行情数据：统一供应商、统一复权方式 │
│ ├── 财务数据：统一报告期处理 │
│ ├── 停牌处理：统一规则（剔除 / 持仓 / 前收盘价） │
│ └── 涨跌停处理：统一规则（能买到吗？能卖掉吗？） │
│ │
│ 3. 统一绩效指标 │
│ ├── 收益指标：年化、超额、对冲 │
│ ├── 风险指标：波动率、最大回撤、VaR │
│ ├── 风险调整：Sharpe、Sortino、Calmar │
│ └── 交易指标：换手率、交易成本占比、容量 │
│ │
└────────────────────────────────────────────────────────┘
4.3 回测代码审查清单
每次提交回测结果时，需要经过以下审查：
| 类别 | 检查项 | 说明 |
|---|---|---|
| 数据 | 是否使用了幸存者偏差数据？ | 数据是否包含退市股票 |
| 数据 | 是否存在前视偏差？ | 信号计算是否只用了当时可得的信息 |
| 数据 | 停牌股票如何处理？ | 停牌期间是否正确处理 |
| 成本 | 交易成本模型是否合理？ | 滑点是否按交易金额动态计算 |
| 成本 | 换手率是否合理？ | 月换手率 > 500% 通常有问题 |
| 过拟合 | 参数数量是否过多？ | 参数越多，过拟合风险越大 |
| 过拟合 | 是否做了样本外验证？ | 至少需要 30% 数据做样本外 |
| 过拟合 | 是否做了稳健性检验？ | 改变参数/时间窗口/股票池后的结果 |
| 复现 | 代码是否可独立运行？ | 其他人能否复现结果 |
| 复现 | 随机种子是否固定？ | 涉及随机过程时需要固定种子 |
| 逻辑 | 是否检查了极端情况？ | 涨跌停、新股、ST 股票 |
| 逻辑 | 多空对冲是否正确？ | 真的是市场中性吗 |
4.4 多层验证体系
机构不会因为一个回测好看就上线。验证是分层的：
多层验证体系：
第一层：样本内验证 (In-Sample)
─────────────────────────────
- 用全部数据拟合参数
- 优化策略参数
- 得到"最好情况"的结果
- 目的：验证策略是否有基本面的有效性
- 注意：样本内结果不能用于上线决策
第二层：样本外验证 (Out-of-Sample)
─────────────────────────────
- 留出 30%-50% 数据完全不参与任何决策
- 包括：参数选择、特征选择、模型选择
- 理想做法：Walk-forward 验证（滚动窗口）
- 目的：验证策略在新数据上的表现
- 标准：样本外 Sharpe > 样本内 Sharpe 的 50%
第三层：模拟盘 (Paper Trading)
─────────────────────────────
- 用实时数据运行策略
- 但不下真实订单
- 记录"如果交易"的结果
- 目的：验证执行层面的可行性
- 时间：至少运行 3-6 个月
- 注意：模拟盘不考虑市场冲击
第四层：实盘 (Live Trading)
─────────────────────────────
- 用真实资金交易
- 从小资金开始（比如目标容量的 10%）
- 逐步加仓
- 目的：验证真实执行效果
- 关键指标：滑点 vs 预期、实际换手率 vs 回测
每一层都可能发现问题，发现问题就回到上一层修复
4.5 回测结果的可复现性要求
| 要求 | 说明 |
|---|---|
| 固定随机种子 | 任何涉及随机过程（蒙特卡洛、Bootstrap）的代码必须固定种子 |
| 记录完整环境 | Python 版本、库版本、操作系统、数据版本 |
| 独立可运行 | 给代码和数据，任何人都能跑出完全相同的结果（精确到小数点后 6 位） |
| 版本控制 | 所有代码纳入 Git 管理，每次回测打 Tag |
| 结果存档 | 回测结果（绩效指标、交易记录、持仓快照）永久存档 |
4.6 Python：标准化回测框架示例
import numpy as np
import pandas as pd
from dataclasses import dataclass
from typing import Optional
# ============================================================
# 标准化回测框架
# ============================================================
@dataclass
class BacktestConfig:
"""回测配置（标准化参数）"""
# 数据
data_source: str = "unified_warehouse"
data_version: str = "2024Q4"
include_delisted: bool = True # 包含退市股票
# 成本模型
commission_rate: float = 0.0003 # 佣金万三
stamp_tax_rate: float = 0.001 # 印花税千一（卖出）
slippage_model: str = "linear" # 滑点模型
slippage_bps: float = 5.0 # 基础滑点 5bp
# 约束
max_turnover_monthly: float = 5.0 # 月换手上限 500%
max_position_pct: float = 0.05 # 单只股票最大持仓 5%
min_liquidity: float = 1e7 # 最低流动性（成交额）
# 验证
train_ratio: float = 0.7 # 样本内比例
random_seed: int = 42
@dataclass
class BacktestResult:
"""回测结果"""
# 收益指标
annual_return: float
annual_excess_return: float
annual_volatility: float
# 风险调整
sharpe: float
sortino: float
calmar: float
max_drawdown: float
# 交易指标
annual_turnover: float
cost_ratio: float # 交易成本占总收益的比例
# 容量
estimated_capacity: float
# 检查
is_overfitted: bool
reproducible: bool
class StandardBacktest:
"""标准化回测引擎"""
def __init__(self, config: BacktestConfig):
self.config = config
def run(self, signal: pd.DataFrame, prices: pd.DataFrame) -> dict:
"""执行回测并返回结果"""
np.random.seed(self.config.random_seed)
n_days = len(signal)
n_stocks = signal.shape[1]
# 模拟回测
returns = self._simulate_trading(signal, prices)
# 计算绩效
result = self._calculate_metrics(returns)
# 过拟合检查
result['is_overfitted'] = self._check_overfitting(signal)
result['reproducible'] = True # 标准化框架默认可复现
return result
def _simulate_trading(self, signal, prices):
"""模拟交易过程"""
# 简化版：用信号的多空组合收益
long_mask = signal > signal.quantile(0.8, axis=1, keepdim=False).values[:, 0]
short_mask = signal < signal.quantile(0.2, axis=1, keepdim=False).values[:, 0]
n_stocks = signal.shape[1]
long_returns = np.where(long_mask[:, np.newaxis].repeat(n_stocks, axis=1),
prices.pct_change().fillna(0).values,
0)
short_returns = np.where(short_mask[:, np.newaxis].repeat(n_stocks, axis=1),
-prices.pct_change().fillna(0).values,
0)
portfolio_returns = (long_returns + short_returns) / 20 # 等权 20 只
daily_returns = portfolio_returns.sum(axis=1)
# 扣除交易成本
turnover = np.mean(np.abs(np.diff(signal.values, axis=0)), axis=1)
cost = turnover * self.config.slippage_bps / 10000
return pd.Series(daily_returns[1:] - cost[:len(daily_returns)-1])
def _calculate_metrics(self, returns):
"""计算标准化绩效指标"""
annual_factor = 252
total_return = (1 + returns).prod() - 1
n_years = len(returns) / annual_factor
annual_return = (1 + total_return) ** (1 / n_years) - 1
annual_vol = returns.std() * np.sqrt(annual_factor)
sharpe = annual_return / annual_vol if annual_vol > 0 else 0
cumulative = (1 + returns).cumprod()
rolling_max = cumulative.cummax()
drawdown = (cumulative - rolling_max) / rolling_max
max_drawdown = drawdown.min()
return {
'annual_return': annual_return,
'annual_volatility': annual_vol,
'sharpe': sharpe,
'max_drawdown': max_drawdown,
'total_return': total_return,
}
def _check_overfitting(self, signal):
"""检查过拟合指标"""
# 简化版：信号的自相关太高可能意味着过拟合
avg_autocorr = np.mean([
pd.Series(signal.iloc[:, i]).autocorr(lag=1)
for i in range(min(10, signal.shape[1]))
if pd.Series(signal.iloc[:, i]).std() > 0
])
return abs(avg_autocorr) > 0.5
# ============================================================
# 回测结果审查
# ============================================================
class BacktestReviewer:
"""回测审查器"""
def review(self, result: dict) -> dict:
"""审查回测结果"""
issues = []
# 1. Sharpe 过高
if result['sharpe'] > 3.0:
issues.append(f"Sharpe = {result['sharpe']:.2f} 过高，可能存在过拟合")
# 2. 最大回撤过小
if result['max_drawdown'] > -0.05:
issues.append(f"最大回撤仅 {result['max_drawdown']:.2%}，可能低估了风险")
# 3. 过拟合检查
if result['is_overfitted']:
issues.append("过拟合检测未通过")
return {
'passed': len(issues) == 0,
'issues': issues,
'verdict': '通过审查' if len(issues) == 0 else
'需要修正' if len(issues) <= 2 else '不建议推进'
}
# 使用示例
print("=" * 60)
print("标准化回测审查示例")
print("=" * 60)
config = BacktestConfig()
bt = StandardBacktest(config)
# 模拟信号和价格
np.random.seed(42)
n_days, n_stocks = 500, 100
signal = pd.DataFrame(np.random.normal(0, 1, (n_days, n_stocks)))
prices = pd.DataFrame(
100 + np.cumsum(np.random.normal(0, 1, (n_days, n_stocks)), axis=0)
)
result = bt.run(signal, prices)
print(f"\n回测绩效:")
for k, v in result.items():
if isinstance(v, float):
print(f" {k}: {v:.4f}")
else:
print(f" {k}: {v}")
reviewer = BacktestReviewer()
review = reviewer.review(result)
print(f"\n审查结果: {review['verdict']}")
if review['issues']:
print("问题列表:")
for issue in review['issues']:
print(f" - {issue}")五、Gate 治理（Gate Governance）
5.1 核心思想：策略上线不是”研究员说了算”
这是治理体系最核心的一环。Gate 治理确保只有通过严格验证的策略才能进入实盘。
白话版本：Gate 就是一道”闸门”，不合格的策略被挡在门外。
为什么需要 Gate？因为一个失败的策略上线后，亏损的不只是策略本身——它会侵蚀整个组合的收益，占用交易资源，分散研究员的精力。
Gate 治理的层次：
研究员 Gate 检查 投资委员会
┌───────────┐ ┌───────────────┐ ┌──────────────┐
│ │ │ │ │ │
│ "我的策略 │ ───────→ │ 自动化检查 │ ───────→ │ 最终决策 │
│ Sharpe 3 │ │ │ │ │
│ 太好了！" │ │ Sharpe > 1.5? │ │ 通过/不通过 │
│ │ │ IC > 0.03? │ │ 资金分配 │
└───────────┘ │ 换手率 < 500%?│ │ │
│ 容量 > 5000万?│ └──────────────┘
│ 相关性 < 0.3? │
│ │
│ 全部通过 → │
│ 部分通过 → │
│ 不通过 → │
└───────────────┘
5.2 策略上线的 Gate 检查点
| 指标 | 最低标准 | 理想标准 | 说明 |
|---|---|---|---|
| 样本外 Sharpe | > 1.0 | > 1.5 | 年化，扣除全部交易成本后 |
| 样本外 IC | > 0.02 | > 0.05 | 月度 IC 均值 |
| 样本外 ICIR | > 0.5 | > 1.0 | IC 的稳定性 |
| 月换手率 | < 800% | < 300% | 过高换手意味着成本吞噬收益 |
| 最大回撤 | < 30% | < 15% | 风险控制底线 |
| 估计容量 | > 2000 万 | > 1 亿 | 考虑市场流动性 |
| 与现有策略相关性 | < 0.5 | < 0.3 | 高相关性意味着分散化不足 |
| 回撤恢复期 | < 12 个月 | < 6 个月 | 回撤后能恢复吗 |
| 盈利月占比 | > 55% | > 60% | 大多数月份应该是赚钱的 |
5.3 实盘监控与熔断机制
策略上线后不是”放任自流”。机构有严格的监控和熔断机制：
实盘监控与熔断：
正常运行 ──→ 触发告警 ──→ 触发熔断 ──→ 退市评估
│ │ │ │
↓ ↓ ↓ ↓
定期 指标偏离 自动减仓 投资委员会
报告 预设阈值 或清仓 决定是否
（周/月） （持续监控） （保护资金） 恢复或退市
常见告警阈值：
┌──────────────────────────────────────────────────┐
│ 告警级别 触发条件 动作 │
├──────────────────────────────────────────────────┤
│ 绿灯 所有指标正常 继续运行 │
│ 黄灯 单日亏损 > 2 个标准差 发送告警通知 │
│ 橙灯 连续 5 日亏损 降仓 50% │
│ 红灯 月度亏损 > 5% 停止交易 │
│ 黑灯 月度亏损 > 10% 清仓 + 紧急审查│
└──────────────────────────────────────────────────┘
5.4 策略退市标准与流程
策略退市的触发条件：
1. IC 持续衰减
- 连续 6 个月滚动 IC < 0.01
- 连续 3 个月滚动 ICIR < 0.3
2. 持续亏损
- 连续 3 个月实盘亏损
- 累计亏损超过初始资本的 10%
3. 拥挤度上升
- 与已知公开因子的相关性 > 0.8
- 因子收益的波动率显著上升
4. 容量急剧下降
- 可交易容量 < 初始估计的 50%
- 冲击成本 > 预期的 2 倍
退市流程：
┌─────────────────────────────────────────────┐
│ 触发退市条件 │
│ ↓ │
│ 自动化系统发出退市建议 │
│ ↓ │
│ 研究员进行原因分析（2 周内） │
│ ↓ │
│ 投资委员会评审 │
│ ↓ │
│ 决策：立即退市 / 观察期 / 修正后恢复 │
│ ↓ │
│ 执行：减仓计划（避免一次性清仓的冲击） │
│ ↓ │
│ 复盘：为什么失效？有什么教训？ │
└─────────────────────────────────────────────┘
5.5 多策略组合的相关性管理
多策略相关性管理：
┌─────────────────────────────────────────────────┐
│ 组合层面治理 │
├─────────────────────────────────────────────────┤
│ │
│ 策略 A (动量) ──┐ │
│ 策略 B (价值) ──┼──→ 相关性矩阵 │
│ 策略 C (低波动) ──┘ │ │
│ ↓ │
│ ┌───────────────┐ │
│ │ 相关性检查： │ │
│ │ │ │
│ │ A-B: 0.1 OK │ │
│ │ A-C: 0.6 !! │ │
│ │ B-C: 0.2 OK │ │
│ │ │ │
│ │ 问题：A 和 C │ │
│ │ 高度相关， │ │
│ │ 分散化不足 │ │
│ └───────┬───────┘ │
│ ↓ │
│ 决策： │
│ 1. 降低 A 或 C 的权重 │
│ 2. 修改 C 减少与 A 的相关 │
│ 3. 寻找替代策略 │
│ │
└─────────────────────────────────────────────────┘
5.6 Python：Gate 检查系统
import numpy as np
import pandas as pd
from dataclasses import dataclass
from typing import List
# ============================================================
# Gate 治理检查系统
# ============================================================
@dataclass
class GateThreshold:
"""Gate 检查阈值"""
min_sharpe: float = 1.0
min_ic: float = 0.02
min_icir: float = 0.5
max_turnover: float = 8.0 # 年化换手率倍数
max_drawdown: float = -0.30
min_capacity: float = 2e7 # 2000 万
max_correlation: float = 0.5
min_win_rate: float = 0.55
@dataclass
class GateCheckResult:
"""单项检查结果"""
name: str
value: float
threshold: float
passed: bool
severity: str # "critical", "warning", "pass"
class GateGovernance:
"""Gate 治理系统"""
def __init__(self, threshold: GateThreshold = None):
self.threshold = threshold or GateThreshold()
def check_all(self, strategy_metrics: dict,
existing_correlations: List[float] = None) -> dict:
"""执行全部 Gate 检查"""
t = self.threshold
checks = []
# Sharpe
checks.append(GateCheckResult(
name="样本外 Sharpe",
value=strategy_metrics['sharpe'],
threshold=t.min_sharpe,
passed=strategy_metrics['sharpe'] >= t.min_sharpe,
severity="critical"
))
# IC
checks.append(GateCheckResult(
name="样本外 IC",
value=strategy_metrics['ic'],
threshold=t.min_ic,
passed=strategy_metrics['ic'] >= t.min_ic,
severity="critical"
))
# ICIR
checks.append(GateCheckResult(
name="样本外 ICIR",
value=strategy_metrics['icir'],
threshold=t.min_icir,
passed=strategy_metrics['icir'] >= t.min_icir,
severity="critical"
))
# 换手率
checks.append(GateCheckResult(
name="年化换手率",
value=strategy_metrics['turnover'],
threshold=t.max_turnover,
passed=strategy_metrics['turnover'] <= t.max_turnover,
severity="warning"
))
# 最大回撤
checks.append(GateCheckResult(
name="最大回撤",
value=strategy_metrics['max_drawdown'],
threshold=t.max_drawdown,
passed=strategy_metrics['max_drawdown'] >= t.max_drawdown,
severity="critical"
))
# 容量
checks.append(GateCheckResult(
name="估计容量",
value=strategy_metrics['capacity'],
threshold=t.min_capacity,
passed=strategy_metrics['capacity'] >= t.min_capacity,
severity="warning"
))
# 盈利月占比
checks.append(GateCheckResult(
name="盈利月占比",
value=strategy_metrics['win_rate'],
threshold=t.min_win_rate,
passed=strategy_metrics['win_rate'] >= t.min_win_rate,
severity="warning"
))
# 与现有策略的相关性
if existing_correlations:
max_corr = max(existing_correlations)
checks.append(GateCheckResult(
name="与现有策略最大相关性",
value=max_corr,
threshold=t.max_correlation,
passed=max_corr <= t.max_correlation,
severity="critical"
))
# 汇总结果
critical_fails = [c for c in checks
if not c.passed and c.severity == "critical"]
warning_fails = [c for c in checks
if not c.passed and c.severity == "warning"]
all_passed = len(critical_fails) == 0
if all_passed and len(warning_fails) == 0:
decision = "PASS"
message = "通过所有 Gate 检查，可以进入投资委员会评审"
elif all_passed:
decision = "PASS_WITH_WARNING"
message = f"通过关键检查，但有 {len(warning_fails)} 项警告需讨论"
else:
decision = "FAIL"
message = f"{len(critical_fails)} 项关键检查未通过，建议重新研究"
return {
'decision': decision,
'message': message,
'checks': checks,
'critical_fails': critical_fails,
'warning_fails': warning_fails,
}
# ============================================================
# 使用示例
# ============================================================
np.random.seed(42)
gate = GateGovernance()
# 模拟两个策略的 Gate 检查
strategies = [
{
'name': '策略A: 高质量多因子',
'metrics': {
'sharpe': 1.8,
'ic': 0.045,
'icir': 1.2,
'turnover': 3.5,
'max_drawdown': -0.12,
'capacity': 5e7,
'win_rate': 0.62,
},
'correlations': [0.15, 0.22, 0.08]
},
{
'name': '策略B: 日内反转',
'metrics': {
'sharpe': 2.5, # 看起来很好
'ic': 0.065,
'icir': 1.8,
'turnover': 15.0, # 但换手率极高
'max_drawdown': -0.08,
'capacity': 5000000, # 容量极小
'win_rate': 0.55,
},
'correlations': [0.6, 0.45, 0.55] # 与现有策略高度相关
},
]
for strategy in strategies:
print(f"\n{'='*60}")
print(f"Gate 检查: {strategy['name']}")
print(f"{'='*60}")
result = gate.check_all(
strategy['metrics'],
strategy['correlations']
)
print(f"\n检查结果:")
for check in result['checks']:
status = "PASS" if check.passed else "FAIL"
if check.name in ("估计容量",):
print(f" [{status}] {check.name}: "
f"{check.value/1e7:.1f} 亿 vs {check.threshold/1e7:.1f} 亿")
else:
print(f" [{status}] {check.name}: "
f"{check.value:.3f} vs {check.threshold:.3f}")
print(f"\n决策: {result['decision']}")
print(f"说明: {result['message']}")六、治理体系的设计原则
6.1 治理不是限制创新，而是防止”假 Alpha”
这是一个需要反复强调的观点。
治理的真正目的：
错误理解："治理就是层层审批，拖慢研究速度"
正确理解："治理是一个过滤器，
它让好想法通过、坏想法被识别，
最终节省的是所有人的时间"
┌─────────────────────────────────────────────┐
│ │
│ 没有治理的研究团队： │
│ - 研究员花 3 个月开发策略 │
│ - 回测看起来很好 │
│ - 上线后亏钱 │
│ - 又花 3 个月排查原因 │
│ - 发现是数据问题 │
│ - 总计浪费 6 个月 │
│ │
│ 有治理的研究团队： │
│ - 研究员花 1 个月开发策略 │
│ - Gate 检查发现数据问题 │
│ - 修正后重新验证 │
│ - 通过 Gate 后上线 │
│ - 实盘表现符合预期 │
│ - 总计 2 个月（其中治理只多花了 1 周） │
│ │
└─────────────────────────────────────────────┘
6.2 从个人研究到机构研究的转变
| 维度 | 个人研究者 | 小团队（3-5 人） | 大型机构（20+ 人） |
|---|---|---|---|
| 命题管理 | 用笔记记录即可 | 共享文档 + 简单评审 | 完整的命题管理系统 |
| 数据治理 | 检查清单 | 自动化质量检查 | 数据仓库 + 版本控制 + 血缘追踪 |
| 特征治理 | 代码注释 | 命名规范 + 文档 | 特征平台 + 自动化管道 |
| 回测治理 | 标准化模板 | 统一回测框架 | 多层验证 + 同行评审 |
| Gate 治理 | 简单的检查清单 | 固定的 Gate 阈值 | 动态阈值 + 投资委员会 |
关键原则：治理的复杂度要与团队规模匹配。 个人研究者不需要机构的完整流程，但必须理解其中的原理。
最小可用治理（个人研究者）：
1. 命题记录本
- 每个想法写下来（动机、经济学逻辑）
- 记录验证结果（通过/不通过/原因）
2. 数据检查清单
- 是否包含退市股票？
- 是否存在前视偏差？
- 停牌股票如何处理？
- 换手率和成本是否合理？
3. 样本外验证
- 至少留出 30% 数据做测试
- 不要在样本外数据上调参
4. 回测复现检查
- 固定随机种子
- 记录环境信息
- 能独立复现结果
6.3 文档化、流程化、自动化
治理体系演进的三个阶段：
治理体系成熟度模型：
Level 1: 文档化
─────────────────
- 有了检查清单
- 有了模板
- 有了规范文档
- 但依赖人工执行
Level 2: 流程化
─────────────────
- 检查变成了流程的一部分
- 有明确的审批流程
- 有定期的评审机制
- 但仍有大量手动操作
Level 3: 自动化
─────────────────
- 数据质量自动检查
- 回测框架自动标准化
- Gate 检查自动执行
- 监控告警自动触发
- 人工只做最终决策
6.4 工具链与平台支持
| 功能 | 工具/平台 | 说明 |
|---|---|---|
| 版本控制 | Git | 所有代码必须纳入版本控制 |
| 数据版本控制 | DVC / LakeFS | 数据文件的版本管理 |
| 实验追踪 | MLflow / W&B | 记录每次实验的参数、结果、环境 |
| 任务调度 | Airflow / Prefect | 自动化数据管道和回测任务 |
| 文档管理 | Confluence / Notion | 研究文档的集中管理 |
| 监控告警 | Grafana + Prometheus | 实盘监控和告警 |
| 代码审查 | GitHub PR / GitLab MR | 同行评审流程 |
七、小结：治理体系的完整图景
五大治理体系的关系
研究治理完整图景：
┌─────────────────────────────────────────────────────────────┐
│ 研究治理体系 │
├─────────────────────────────────────────────────────────────┤
│ │
│ 输入：想法（灵感、文献、市场观察） │
│ ↓ │
│ ┌─────────────┐ │
│ │ 命题治理 │ 想法值得研究吗？经济学逻辑成立吗？ │
│ └──────┬──────┘ │
│ ↓ │
│ ┌─────────────┐ │
│ │ 数据治理 │ 数据干净吗？没有偏差吗？ │
│ └──────┬──────┘ │
│ ↓ │
│ ┌─────────────┐ │
│ │ 特征治理 │ 特征可靠吗？可复现吗？在衰减吗？ │
│ └──────┬──────┘ │
│ ↓ │
│ ┌─────────────┐ │
│ │ 回测治理 │ 回测结果可信吗？成本合理吗？ │
│ └──────┬──────┘ │
│ ↓ │
│ ┌─────────────┐ │
│ │ Gate 治理 │ 策略准备好上线了吗？通过全部检查了吗？ │
│ └──────┬──────┘ │
│ ↓ │
│ 输出：可信的投资决策 │
│ ↓ │
│ 持续监控 → 退市评估 → 新的命题 │
│ │
└─────────────────────────────────────────────────────────────┘
核心要点总结
| 治理领域 | 核心问题 | 关键工具 | 最常见的失误 |
|---|---|---|---|
| 命题治理 | 想法值得研究吗？ | 命题评审表、优先级矩阵 | 纯数据驱动，没有经济学逻辑 |
| 数据治理 | 数据可信吗？ | 质量检查框架、偏差检测 | 幸存者偏差和前视偏差 |
| 特征治理 | 特征可靠吗？ | 特征平台、监控仪表盘 | 特征没有文档化，无法复现 |
| 回测治理 | 回测结果可信吗？ | 标准化框架、审查清单 | 过拟合、低估交易成本 |
| Gate 治理 | 策略能上线吗？ | Gate 检查系统、熔断机制 | 跳过 Gate 直接上线 |
最后的忠告：
治理的目的是让你在实盘赚钱，而不是让你在流程中迷失。好的治理应该像一个自动扶梯——你站上去就自然被带到正确的方向，而不是像爬山一样每一步都艰难。
如果你的治理流程让研究员觉得”我在填表格而不是在做研究”，那说明治理设计有问题。最有效的治理，是嵌入到工具和流程中的，而不是依赖人的自觉性的。
推荐延伸阅读
| 书名 | 作者 | 相关章节 |
|---|---|---|
| Advances in Financial Machine Learning | Marcos Lopez de Prado | 第 1-4 章（数据治理、回测陷阱、特征工程） |
| Quantitative Portfolio Management | Michael Isichenko | 第 8-10 章（研究流程、风险管理） |
| The Man Who Solved the Market | Gregory Zuckerman | 文艺作品，但生动展现了文艺复兴科技的研究文化 |
| Expected Returns | Antti Ilmanen | 第 2 章（投资哲学）和各因子章节 |
版本信息
- 创建日期：2026-03-28
- 最后更新：2026-03-28
- 许可：CC BY-NC-SA 4.0