14 KiB
14 KiB
框架通用能力与定制开发边界设计
一、框架设计原则
1.1 核心原则
框架应该提供：
- ✅ 所有策略共用的基础设施（数据获取、回测引擎、风控组件）
- ✅ 可扩展的抽象接口（因子、信号、策略）
- ✅ 通用的计算工具（技术指标、绩效指标）
框架不应该提供：
- ❌ 特定策略的业务逻辑（分散化选股、溢价过滤阈值）
- ❌ 特定市场的数据适配（A股交易日历对齐、指数-ETF映射）
- ❌ 特定资产的交易规则（加密货币7x24交易、期货保证金）
二、通用能力与定制开发划分
2.1 框架通用能力（Framework Core）
| 模块 | 通用能力 | 说明 |
|---|---|---|
| 数据层 | DataSource抽象接口 | 定义数据获取标准接口 |
| OHLCV数据结构 | 统一的K线数据格式 | |
| 缓存管理 | 本地缓存、版本控制 | |
| 数据验证 | 数据完整性检查 | |
| 因子层 | FactorBase抽象类 | 因子计算接口 |
| FactorRegistry注册器 | 因子管理机制 | |
| FactorCombiner组合器 | 多因子加权组合 | |
| 技术指标库（通用） | RSI/MACD/MA/ATR/Bollinger | |
| 动量因子（通用） | 加权动量、简单动量 | |
| 信号层 | SignalGenerator抽象类 | 信号生成接口 |
| 基础信号生成器 | TopNSelector、ThresholdSelector | |
| 风控层 | RiskControl抽象类 | 风控组件接口 |
| 通用风控组件 | StopLoss、PositionLimit、TrailingStop | |
| CallbackHook回调机制 | 生命周期钩子 | |
| 执行层 | Executor抽象类 | 执行器接口 |
| BacktestExecutor回测引擎 | 净值计算、交易记录 | |
| DryRunExecutor模拟盘 | 模拟交易执行 | |
| Portfolio持仓管理 | 持仓数据结构 | |
| 报告层 | ReportGenerator | 绩效指标计算 |
| 通用可视化 | 净值曲线、收益分布 | |
| 配置层 | ConfigLoader | YAML配置加载 |
| 配置验证 | 必填项检查 |
2.2 定制开发部分（Strategy Specific）
| 模块 | 定制开发内容 | 所属策略 |
|---|---|---|
| 数据层 | A股交易日历对齐 | 跨市场策略 |
| 指数-ETF映射管理 | ETF轮动策略 | |
| SSH隧道配置 | 网络受限环境 | |
| 溢价率数据获取 | 跨境ETF策略 | |
| 因子层 | 分散化选股因子 | 轮动策略 |
| 崩盘过滤（特定阈值） | 动量策略 | |
| 动态ATR周期（特定参数） | 自适应策略 | |
| 信号层 | 分散化选股逻辑 | 轮动策略 |
| 调仓阈值检查 | 轮动策略 | |
| 超买超卖信号（特定阈值） | 反转策略 | |
| 风控层 | 溢价过滤（特定阈值） | 跨境ETF策略 |
| 崩盘检测（特定阈值） | 动量策略 | |
| 持仓时间动态止损 | 特定策略 | |
| 策略层 | 策略业务逻辑 | 每个策略不同 |
| 候选池管理 | 每个策略不同 | |
| 调仓频率控制 | 每个策略不同 |
三、详细设计方案
3.1 框架核心目录结构（通用能力）
framework/
├── core/ # 框架核心（所有策略共用）
│ ├── data/ # 数据层核心
│ │ ├── base.py # DataSource抽象接口
│ │ ├── ohlcv.py # OHLCV数据结构
│ │ ├── cache.py # 缓存管理（版本控制）
│ │ └── validator.py # 数据验证
│ │
│ ├── factors/ # 因子层核心
│ │ ├── base.py # FactorBase抽象
│ │ ├── registry.py # FactorRegistry
│ │ ├── combiner.py # FactorCombiner
│ │ └── indicators/ # 通用技术指标库
│ │ │ ├── momentum.py # 动量因子（通用）
│ │ │ ├── trend.py # 趋势因子（通用）
│ │ │ ├── volatility.py# 波动率因子（通用）
│ │ │ └── oscillator.py# 震荡指标（RSI/KDJ）
│ │
│ ├── signals/ # 信号层核心
│ │ ├── base.py # SignalGenerator抽象
│ │ ├── selectors/ # 基础选股器
│ │ │ ├── top_n.py # Top N选股（通用）
│ │ │ └ threshold.py # 阈值选股（通用）
│ │ │
│ ├── risk/ # 风控层核心
│ │ ├── base.py # RiskControl抽象
│ │ ├── controls/ # 通用风控组件
│ │ │ ├── stop_loss.py # 止损控制（通用）
│ │ │ ├── position.py # 仓位控制（通用）
│ │ │ └ trailing.py # 跟踪止损（通用）
│ │ ├── callback.py # CallbackHook机制
│ │
│ ├── execution/ # 执行层核心
│ │ ├── base.py # Executor抽象
│ │ ├── backtest.py # 完整回测引擎
│ │ ├── dry_run.py # 模拟盘执行
│ │ ├── portfolio.py # 持仓管理
│ │
│ ├── report/ # 报告层核心
│ │ ├── generator.py # 报告生成器
│ │ ├── metrics.py # 绩效指标（Sharpe/MaxDD/CAGR）
│ │ ├── visualizer.py # 可视化（净值曲线、收益分布）
│ │
│ └── config/ # 配置层核心
│ ├── loader.py # ConfigLoader
│ ├── validator.py # 配置验证
3.2 定制开发目录结构（策略特定）
strategies/
├── rotation/ # 轮动策略（定制）
│ ├── strategy.py # 策略业务逻辑
│ ├── factors.py # 分散化因子
│ ├── signals.py # 分散化选股逻辑
│ ├── risk.py # 溢价过滤、崩盘检测
│ ├── data_adapter.py # A股交易日历对齐
│ ├── etf_mapper.py # 指数-ETF映射管理
│ ├── config.yaml # 策略配置
│
├── trend_follow/ # 趋势跟踪策略（定制）
│ ├── strategy.py # 策略业务逻辑
│ ├── factors.py # 趋势强度因子
│ ├── signals.py # 趋势跟随信号
│ ├── risk.py # 趋势止损规则
│ ├── config.yaml # 策略配置
│
├── reversal/ # 反转策略（定制）
│ ├── strategy.py # 策略业务逻辑
│ ├── factors.py # 超买超卖因子
│ ├── signals.py # 反转信号
│ ├── risk.py # 反转止损规则
│ ├── config.yaml # 策略配置
│
└── shared/ # 多策略共享的定制组件
├── data_sources/ # 定制数据源
│ ├── tushare.py # Tushare数据源
│ ├── yfinance.py # YFinance数据源
│ ├── hybrid.py # 混合数据源
│ ├── ssh_tunnel.py # SSH隧道管理
│
├── market_adapters/ # 市场适配器
│ ├── a_share_calendar.py # A股交易日历
│ ├── cross_market_align.py# 跨市场对齐
│
└── utils/ # 定制工具
├── premium_calculator.py# 溢价率计算
├── etf_mapper.py # ETF映射管理
四、接口设计示例
4.1 框架核心接口（通用）
# framework/core/data/base.py
class DataSource(ABC):
"""数据源抽象接口（通用）"""
@abstractmethod
def fetch_ohlcv(self, code: str, start: str, end: str) -> pd.DataFrame:
"""获取OHLCV数据"""
pass
@abstractmethod
def get_supported_codes(self) -> List[str]:
"""获取支持的代码列表"""
pass
# framework/core/factors/indicators/momentum.py
class MomentumFactor(FactorBase):
"""动量因子（通用）"""
name = "momentum"
def compute(self, data: pd.DataFrame) -> pd.Series:
"""计算动量（通用计算逻辑）"""
prices = data['close']
return prices.pct_change(self.params['n_days'])
# framework/core/signals/selectors/top_n.py
class TopNSelector(SignalGenerator):
"""Top N选股器（通用）"""
mode = "top_n"
def generate(self, factor_data: pd.DataFrame) -> pd.DataFrame:
"""按因子值排序选Top N（通用逻辑）"""
scores = factor_data[self.factor_cols]
top_n = scores.apply(lambda row: row.nlargest(self.select_num).index.tolist(), axis=1)
return pd.DataFrame({'signal': top_n})
# framework/core/risk/controls/stop_loss.py
class StopLossControl(RiskControl):
"""止损控制（通用）"""
name = "stop_loss"
def check(self, position: Position) -> bool:
"""检查是否触发止损（通用逻辑）"""
return position.profit_ratio > self.threshold
4.2 定制开发示例（策略特定）
# strategies/rotation/data_adapter.py
class AShareCalendarAdapter:
"""A股交易日历适配器（定制：仅A股策略需要）"""
def align_to_a_share_calendar(self, data: pd.DataFrame) -> pd.DataFrame:
"""对齐到A股交易日历"""
# 从Tushare获取A股交易日历
a_share_dates = self.get_a_share_trading_dates()
return data.reindex(a_share_dates, method='ffill')
# strategies/rotation/etf_mapper.py
class ETFMapper:
"""指数-ETF映射管理（定制：仅ETF轮动策略需要）"""
def __init__(self, mapping_config: dict):
self.mapping = mapping_config # {index_code: etf_code}
def get_signal_code(self, index_code: str) -> str:
"""获取信号源代码（指数）"""
return index_code
def get_trade_code(self, index_code: str) -> str:
"""获取交易标的代码（ETF）"""
return self.mapping.get(index_code, {}).get('etf', index_code)
def get_premium(self, index_code: str, date: str) -> float:
"""计算溢价率"""
etf_price = self.get_etf_price(index_code, date)
etf_nav = self.get_etf_nav(index_code, date)
return etf_price / etf_nav - 1
# strategies/rotation/factors.py
class DiversifiedMomentumFactor(FactorBase):
"""分散化动量因子（定制：仅分散化轮动策略需要）"""
name = "diversified_momentum"
def __init__(self, n_days: int, crash_threshold: float = 0.05):
self.n_days = n_days
self.crash_threshold = crash_threshold # 定制阈值
def compute(self, data: pd.DataFrame) -> pd.Series:
"""计算动量并应用崩盘过滤"""
# 使用通用动量计算
momentum = self._compute_momentum(data)
# 应用定制崩盘过滤
momentum = self._apply_crash_filter(momentum, data)
return momentum
def _apply_crash_filter(self, momentum: pd.Series, data: pd.DataFrame) -> pd.Series:
"""崩盘过滤（定制阈值）"""
# 3天跌超过crash_threshold清零
# ...
五、配置分层设计
5.1 框架配置（通用）
# config/framework.yaml（框架通用配置）
framework:
version: 1
execution:
default_mode: "backtest"
report:
default_metrics: ["cagr", "sharpe", "max_dd"]
cache:
enabled: true
version_control: true
5.2 策略配置（定制）
# strategies/rotation/config.yaml（策略定制配置）
strategy:
name: "rotation"
# 定制：候选池配置
code_list:
"399006.SZ":
name: "创业板指"
etf: "159915.SZ"
market: "A"
# 定制：因子参数
factors:
- name: "momentum"
params:
n_days: 25
crash_threshold: 0.05 # 定制阈值
# 定制：选股逻辑
signal:
mode: "diversified_top_n" # 定制信号生成器
select_num: 3
group_by: "market" # 定制：按大类分组
# 定制：风控规则
risk:
- name: "premium_filter"
threshold: 0.10 # 定制阈值
markets: ["HK", "US"] # 定制：仅港股美股
# 定制：数据适配器
data_adapter:
calendar: "a_share" # 定制：A股交易日历
etf_mapping: true # 定制：启用指数-ETF映射
六、总结
6.1 划分原则
| 判断标准 | 框架通用 | 定制开发 |
|---|---|---|
| 是否所有策略都需要 | ✅ 是 | ❌ 否 |
| 是否有固定参数 | ✅ 参数可配置 | ❌ 固定阈值 |
| 是否依赖特定市场 | ✅ 市场无关 | ❌ A股/美股特定 |
| 是否依赖特定资产 | ✅ 资产无关 | ❌ ETF/期货特定 |
| 是否可复用 | ✅ 跨策略复用 | ❌ 单策略专用 |
6.2 框架核心交付物
框架必须提供：
- ✅ DataSource抽象接口（不含具体实现）
- ✅ FactorBase抽象 + 通用技术指标库
- ✅ SignalGenerator抽象 + 基础选股器
- ✅ RiskControl抽象 + 通用风控组件
- ✅ 完整回测引擎 + 报告生成器
策略开发者需要实现：
- ❌ 具体DataSource实现（Tushare/YFinance/Hybrid）
- ❌ 定制因子（分散化因子、特定阈值崩盘过滤）
- ❌ 定制信号生成器（分散化选股、调仓阈值）
- ❌ 定制风控组件（溢价过滤、特定止损规则）
- ❌ 定制数据适配器（A股日历、ETF映射）
6.3 框架设计目标
框架目标：
- 提供完整的基础设施
- 提供清晰的扩展接口
- 策略开发者只需关注业务逻辑
策略开发者工作量：
- 编写策略配置（YAML）
- 实现定制组件（继承框架接口）
- 无需关心回测引擎、报告生成等底层逻辑
文档版本：V1.0 设计时间：2026-05-08 目标：明确框架与定制的边界