A股量化投研一体化框架（技术学习版）
免责声明：本文档仅供技术学习和学术研究参考，不构成任何投资建议、投资咨询或证券交易指导。股市有风险，投资需谨慎。任何基于本文档内容进行的投资决策，均由投资者自行承担全部风险和责任。根据《中华人民共和国证券法》，提供投资建议需持有中国证监会颁发的证券投资咨询执业资格证书。本框架中所有策略、模型、指标均为技术演示用途，不保证其准确性、完整性和时效性，不构成对任何投资收益的承诺或暗示。
一、框架概述
1.1 框架定位
本框架是一个面向A股市场的量化投研技术学习框架，旨在为金融工程、量化投资领域的学习者和研究人员提供一套系统化的分析工具集。框架采用模块化架构设计，将量化投研流程拆解为数据采集、指标计算、策略研究、回测验证、组合管理五大核心环节，每个环节包含若干独立可插拔的功能模块。
重要声明：本框架仅供学习研究用途，不用于实际交易。框架中涉及的所有策略、模型、指标均以技术分析工具框架和学术研究方法论的形式呈现，不构成对任何金融产品的推荐或买卖建议。
1.2 设计目标
- 教育导向：帮助学习者系统理解量化投资方法论，从数据获取到策略回测的完整流程
- 模块化设计：20个独立Skill模块，每个模块职责单一、接口统一，支持独立学习和组合使用
- 学术规范：所有策略描述遵循学术研究方法论标准，包含假设条件、适用范围、局限性说明
- 合规优先：严格遵守中国证券市场相关法律法规，明确技术工具的法律边界
- 可扩展性：支持用户基于框架进行二次开发和学术实验
1.3 框架特点
| 特性 | 说明 |
|---|---|
| 模块化设计 | 20个独立Skill模块，即插即用，松耦合高内聚 |
| 统一接口 | 所有模块遵循相同的输入输出规范，便于组合编排 |
| 五层架构 | 数据层→计算层→策略层→回测层→执行层，层次清晰 |
| A股适配 | 内置T+1、涨跌停、停牌等A股特殊交易规则处理 |
| 合规内嵌 | 全流程合规检查，明确禁止实际交易用途 |
1.4 适用市场
- 沪深主板：上海证券交易所、深圳证券交易所主板上市证券
- 创业板：深圳证券交易所创业板（注册制）
- 科创板：上海证券交易所科创板（注册制）
- 北交所：北京证券交易所
- 港股通：沪深港通机制下的港股标的（仅作数据研究参考）
1.5 20个模块总览
技术指标模块（10个）：
| 编号 | 模块名称 | 类别 | 功能概述 |
|---|---|---|---|
| 1 | 股市时间校准 | 基础设施 | 交易日历、交易时段识别、节假日处理 |
| 2 | KDJ | 震荡指标 | 随机指标，判断超买超卖状态 |
| 3 | MACD | 趋势指标 | 异同移动平均线，判断趋势方向与强度 |
| 4 | 成交量 | 量能指标 | 成交量分析与量价关系研究 |
| 5 | 换手率 | 量能指标 | 换手率计算与活跃度分析 |
| 6 | WR | 震荡指标 | 威廉指标，判断超买超卖 |
| 7 | EXPMA | 趋势指标 | 指数移动平均线，趋势跟踪 |
| 8 | ARBR | 震荡指标 | 人气意愿指标，市场情绪分析 |
| 9 | DPO | 震荡指标 | 去趋势价格振荡器，周期分析 |
| 10 | VR | 震荡指标 | 成交量比率，量价背离分析 |
数据模块（10个）：
| 编号 | 模块名称 | 类别 | 功能概述 |
|---|---|---|---|
| 11 | A股热点新闻 | 资讯数据 | 市场热点新闻采集与情绪分析 |
| 12 | 个股公告 | 资讯数据 | 上市公司公告采集与分类 |
| 13 | 集合竞价 | 行情数据 | 集合竞价数据采集与分析 |
| 14 | 历史K线 | 行情数据 | 个股历史K线数据获取 |
| 15 | 指数K线 | 行情数据 | 市场指数K线数据获取 |
| 16 | 政策法规 | 资讯数据 | 宏观政策与行业法规采集 |
| 17 | 董秘互动 | 资讯数据 | 互动平台问答数据采集 |
| 18 | 券商研报 | 研究数据 | 券商研究报告采集与摘要 |
| 19 | 全球央行货币政策 | 研究数据 | 央行政策动态与利率数据 |
| 20 | 美股热点新闻 | 研究数据 | 美股市场新闻与跨市场关联研究 |
二、系统架构设计
2.1 整体架构
本框架采用经典的五层架构设计，自底向上依次为数据层、计算层、策略层、回测层和执行层。每一层具有明确的职责边界，层与层之间通过统一的数据接口进行通信。
┌─────────────────────────────────────────────────────┐
│ 执行层（Execution） │
│ 组合构建 │ 订单管理 │ 实时监控 │ 风险控制 │
├─────────────────────────────────────────────────────┤
│ 回测层（Backtest） │
│ 回测引擎 │ 绩效评估 │ 稳健性检查 │ 归因分析 │
├─────────────────────────────────────────────────────┤
│ 策略层（Strategy） │
│ 策略模板 │ 事件驱动 │ 多因子模型 │ 信号融合 │
├─────────────────────────────────────────────────────┤
│ 计算层（Calculation） │
│ 趋势指标 │ 震荡指标 │ 量能指标 │ 指标组合引擎 │
├─────────────────────────────────────────────────────┤
│ 数据层（Data） │
│ 行情数据 │ 资讯数据 │ 研究数据 │ 数据管理 │
└─────────────────────────────────────────────────────┘
各层职责说明：
- 数据层：负责从多数据源采集、清洗、标准化各类市场数据，包括行情数据、资讯数据和研究数据，为上层提供统一的高质量数据服务
- 计算层：基于数据层提供的标准化数据，计算各类技术指标，包括趋势类、震荡类和量能类指标，并提供指标组合引擎
- 策略层：运用计算层输出的指标信号，按照学术研究方法论构建策略模板，进行事件驱动研究和多因子建模
- 回测层：对策略层输出的交易信号进行历史数据回测验证，评估策略的绩效表现和稳健性
- 执行层：提供组合构建和订单管理的理论方法论，用于学术研究中的组合优化实验
2.2 20个模块依赖关系图谱
数据模块（底层） 技术指标模块（中层）
┌──────────────┐ ┌──────────────┐
│ 14.历史K线 │──────┐ ┌────→│ 2.KDJ │
│ 15.指数K线 │──────┼────┼────→│ 3.MACD │
│ 13.集合竞价 │──────┤ │ │ 6.WR │
└──────────────┘ │ │ │ 7.EXPMA │
┌──────────────┐ │ │ │ 8.ARBR │
│ 11.A股热点 │──────┤ │ │ 9.DPO │
│ 12.个股公告 │──────┼────┼────→│ 10.VR │
│ 16.政策法规 │──────┤ │ │ 4.成交量 │
│ 17.董秘互动 │──────┤ │ │ 5.换手率 │
└──────────────┘ │ │ └──────────────┘
┌──────────────┐ │ │
│ 18.券商研报 │──────┤ │ ┌──────────────┐
│ 19.央行政策 │──────┼────┼────→│ 1.时间校准 │
│ 20.美股新闻 │──────┘ │ │ (贯穿所有层) │
└──────────────┘ │ └──────────────┘
│
┌──────┴──────┐
│ 策略层/回测层 │
│ 执行层 │
└─────────────┘
依赖关系说明：
- 行情数据模块（13、14、15）是技术指标模块（2-10）的基础数据源
- 资讯数据模块（11、12、16、17）主要服务于策略层的事件驱动研究
- 研究数据模块（18、19、20）服务于宏观因子和多因子研究
- 时间校准模块（1）作为基础设施模块，贯穿所有层级
2.3 统一接口设计规范
所有模块遵循统一的接口设计规范，采用Python协议（Protocol）定义标准接口：
from typing import Protocol, Dict, List, Optional, Any
from dataclasses import dataclass
from datetime import datetime, date
import pandas as pd
@dataclass
class ModuleResult:
"""模块统一输出格式"""
module_name: str # 模块名称
data: pd.DataFrame # 输出数据
metadata: Dict[str, Any] # 元数据（计算参数、时间戳等）
warnings: List[str] # 警告信息
error: Optional[str] # 错误信息
class BaseModule(Protocol):
"""所有模块的基础协议"""
module_name: str
module_version: str
def initialize(self, config: Dict[str, Any]) -> None:
"""模块初始化"""
...
def execute(self, params: Dict[str, Any]) -> ModuleResult:
"""执行模块核心功能"""
...
def validate_params(self, params: Dict[str, Any]) -> bool:
"""参数校验"""
...
def get_metadata(self) -> Dict[str, Any]:
"""获取模块元数据"""
...
class DataModule(BaseModule, Protocol):
"""数据模块扩展协议"""
def fetch(self, symbol: str, start_date: date,
end_date: date) -> ModuleResult:
"""数据获取"""
...
def get_update_frequency(self) -> str:
"""获取数据更新频率"""
...
class IndicatorModule(BaseModule, Protocol):
"""指标模块扩展协议"""
def calculate(self, df: pd.DataFrame,
params: Dict[str, Any]) -> ModuleResult:
"""指标计算"""
...
def get_default_params(self) -> Dict[str, Any]:
"""获取默认参数"""
...
三、数据层（10个数据模块整合）
数据层是整个框架的基石，负责多源数据的采集、清洗、标准化和存储管理。10个数据模块按功能分为三个子系统和一个统一管理层。
3.1 行情数据子系统
行情数据子系统包含历史K线（模块14）、指数K线（模块15）和集合竞价（模块13）三个模块，负责市场行情数据的采集与预处理。
3.1.1 历史K线模块（模块14）
功能概述：获取个股历史K线数据，支持日线、分钟线等多种周期。
数据字段规范：
| 字段名 | 类型 | 说明 |
|---|---|---|
| symbol | str | 证券代码（如 600000.SH） |
| trade_date | date | 交易日期 |
| open | float | 开盘价 |
| high | float | 最高价 |
| low | float | 最低价 |
| close | float | 收盘价 |
| volume | float | 成交量（手） |
| amount | float | 成交额（元） |
| pre_close | float | 前收盘价 |
| change_pct | float | 涨跌幅（%） |
| turnover_rate | float | 换手率（%） |
接口定义：
class HistoricalKLineModule:
"""历史K线数据模块"""
module_name = "historical_kline"
module_version = "1.0.0"
def __init__(self):
self._cache: Dict[str, pd.DataFrame] = {}
def fetch_daily(
self,
symbol: str,
start_date: date,
end_date: date,
adjust_type: str = "hfq" # hfq:后复权, qfq:前复权, None:不复权
) -> ModuleResult:
"""
获取日线数据
参数:
symbol: 证券代码
start_date: 开始日期
end_date: 结束日期
adjust_type: 复权类型
返回:
ModuleResult 包含标准化K线数据
"""
# 实现逻辑：从数据源获取原始数据，进行标准化处理
...
3.1.2 指数K线模块（模块15）
功能概述：获取市场主要指数（上证指数、深证成指、创业板指、科创50、沪深300、中证500、中证1000等）的历史K线数据。
class IndexKLineModule:
"""指数K线数据模块"""
module_name = "index_kline"
module_version = "1.0.0"
# 支持的指数列表
SUPPORTED_INDICES = {
"000001.SH": "上证指数",
"399001.SZ": "深证成指",
"399006.SZ": "创业板指",
"000688.SH": "科创50",
"000300.SH": "沪深300",
"000905.SH": "中证500",
"000852.SH": "中证1000",
"899050.BJ": "北证50",
}
def fetch_index_daily(
self,
index_code: str,
start_date: date,
end_date: date
) -> ModuleResult:
"""获取指数日线数据"""
...
3.1.3 集合竞价模块（模块13）
功能概述：采集A股每个交易日的集合竞价数据，包括早盘集合竞价（9:15-9:25）和尾盘集合竞价（14:57-15:00）。
class CallAuctionModule:
"""集合竞价数据模块"""
module_name = "call_auction"
module_version = "1.0.0"
def fetch_auction_data(
self,
symbol: str,
trade_date: date,
auction_type: str = "opening" # opening:早盘, closing:尾盘
) -> ModuleResult:
"""
获取集合竞价数据
数据字段:
symbol: 证券代码
trade_date: 交易日期
auction_time: 竞价时间点
auction_price: 竞价价格
auction_volume: 竞价量
match_count: 匹配笔数
"""
...
3.2 资讯数据子系统
资讯数据子系统包含A股热点新闻（模块11）、个股公告（模块12）、政策法规（模块16）和董秘互动（模块17）四个模块。
3.2.1 A股热点新闻模块（模块11）
功能概述：采集A股市场热点新闻，进行文本分类和情绪标注，为事件驱动研究提供数据支撑。
class MarketNewsModule:
"""A股热点新闻模块"""
module_name = "market_news"
module_version = "1.0.0"
# 新闻分类标签体系
NEWS_CATEGORIES = [
"宏观经济", "行业政策", "公司新闻", "市场动态",
"资金流向", "IPO/增发", "监管动态", "国际影响"
]
def fetch_hot_news(
self,
news_date: date,
category: Optional[str] = None,
top_n: int = 50
) -> ModuleResult:
"""
获取热点新闻
返回字段:
news_id: 新闻唯一标识
title: 标题
content: 正文（摘要）
publish_time: 发布时间
source: 来源
category: 分类标签
sentiment_score: 情绪得分（-1到1，学术研究用途）
related_symbols: 关联证券代码列表
"""
...
3.2.2 个股公告模块（模块12）
功能概述：采集上市公司公告，按公告类型自动分类，支持关键词检索。
class CompanyAnnouncementModule:
"""个股公告模块"""
module_name = "company_announcement"
module_version = "1.0.0"
# 公告类型分类
ANNOUNCEMENT_TYPES = {
"业绩预告": ["业绩预告", "业绩快报", "年度报告"],
"分红送转": ["利润分配", "转增股本", "分红"],
"重大事项": ["重大合同", "重组", "并购", "股权转让"],
"股权变动": ["减持", "增持", "质押", "解押"],
"监管函件": ["问询函", "关注函", "监管函"],
}
def fetch_announcements(
self,
symbol: Optional[str] = None,
start_date: Optional[date] = None,
ann_type: Optional[str] = None,
keyword: Optional[str] = None
) -> ModuleResult:
"""获取公告数据"""
...
3.2.3 政策法规模块（模块16）
功能概述：采集宏观经济政策和行业法规信息，为宏观因子研究和政策事件分析提供数据。
class PolicyRegulationModule:
"""政策法规模块"""
module_name = "policy_regulation"
module_version = "1.0.0"
def fetch_policies(
self,
policy_type: str = "all", # macro/industry/regulatory/all
start_date: Optional[date] = None,
keyword: Optional[str] = None
) -> ModuleResult:
"""
获取政策法规数据
返回字段:
policy_id: 政策唯一标识
title: 标题
content: 摘要
publish_date: 发布日期
source: 发布机构
policy_type: 类型（宏观/行业/监管）
impact_level: 影响等级评估（学术研究用途）
"""
...
3.2.4 董秘互动模块（模块17）
功能概述：采集上市公司互动平台（如深交所互动易、上交所e互动）的投资者问答数据。
class SecretaryInteractionModule:
"""董秘互动模块"""
module_name = "secretary_interaction"
module_version = "1.0.0"
def fetch_interactions(
self,
symbol: str,
start_date: Optional[date] = None,
question_type: Optional[str] = None
) -> ModuleResult:
"""
获取董秘互动数据
返回字段:
interaction_id: 互动唯一标识
symbol: 证券代码
question: 投资者提问内容
answer: 董秘回复内容
question_date: 提问日期
answer_date: 回复日期
question_type: 问题分类
"""
...
3.3 研究数据子系统
研究数据子系统包含券商研报（模块18）、全球央行货币政策（模块19）和美股热点新闻（模块20）三个模块。
3.3.1 券商研报模块（模块18）
功能概述：采集券商研究报告摘要和评级数据，用于研究方法论学习。
class BrokerResearchModule:
"""券商研报模块"""
module_name = "broker_research"
module_version = "1.0.0"
def fetch_research_reports(
self,
symbol: Optional[str] = None,
broker: Optional[str] = None,
report_type: str = "all", # company/industry/strategy/macro
start_date: Optional[date] = None
) -> ModuleResult:
"""
获取研报数据
注意：研报观点仅代表分析师个人判断，不构成投资建议。
本模块仅用于学术研究中的文本分析和观点聚合方法学习。
返回字段:
report_id: 研报唯一标识
title: 标题
summary: 摘要
broker: 券商名称
analyst: 分析师姓名
publish_date: 发布日期
report_type: 报告类型
rating_change: 评级变动（如有）
"""
...
3.3.2 全球央行货币政策模块（模块19）
功能概述：跟踪全球主要央行（中国人民银行、美联储、欧洲央行、日本央行、英格兰银行）的货币政策动态。
class CentralBankPolicyModule:
"""全球央行货币政策模块"""
module_name = "central_bank_policy"
module_version = "1.0.0"
SUPPORTED_CENTRAL_BANKS = [
"PBOC", # 中国人民银行
"FED", # 美联储
"ECB", # 欧洲央行
"BOJ", # 日本央行
"BOE", # 英格兰银行
]
def fetch_policy_events(
self,
bank: Optional[str] = None,
event_type: str = "all", # rate_decision/minutes/speech/all
start_date: Optional[date] = None
) -> ModuleResult:
"""
获取央行政策事件
返回字段:
event_id: 事件唯一标识
bank: 央行代码
event_type: 事件类型（利率决议/会议纪要/官员讲话）
event_date: 事件日期
title: 标题
summary: 摘要
policy_rate: 当前政策利率
rate_change: 利率变动（基点）
"""
...
3.3.3 美股热点新闻模块（模块20）
功能概述：采集美股市场热点新闻，用于跨市场关联性研究和全球宏观因子分析。
class USMarketNewsModule:
"""美股热点新闻模块"""
module_name = "us_market_news"
module_version = "1.0.0"
def fetch_us_news(
self,
news_date: date,
category: Optional[str] = None,
top_n: int = 30
) -> ModuleResult:
"""
获取美股热点新闻
注意：本模块仅用于跨市场关联性学术研究，
不涉及美股投资建议。
返回字段:
news_id: 新闻唯一标识
title: 标题
content: 摘要
publish_time: 发布时间（UTC）
source: 来源
category: 分类
related_tickers: 关联美股代码
"""
...
3.4 数据统一管理层
数据统一管理层负责协调10个数据模块的协同工作，提供多源适配、质量监控和增量更新能力。
class DataManager:
"""数据统一管理层"""
def __init__(self):
self._modules: Dict[str, BaseModule] = {}
self._quality_rules: List[QualityRule] = []
self._update_log: pd.DataFrame = pd.DataFrame()
def register_module(self, module: BaseModule) -> None:
"""注册数据模块"""
self._modules[module.module_name] = module
def validate_data_quality(self, result: ModuleResult) -> QualityReport:
"""
数据质量校验
校验规则:
- 完整性：检查必要字段是否缺失
- 一致性：检查数据格式是否统一
- 时效性：检查数据时间戳是否合理
- 准确性：检查数值范围是否合理（如价格不能为负）
- 连续性：检查交易日数据是否连续（排除停牌日）
"""
...
def incremental_update(self, module_name: str,
last_update: datetime) -> ModuleResult:
"""增量数据更新，仅获取上次更新后的新数据"""
...
def get_data_snapshot(self, symbols: List[str],
fields: List[str]) -> pd.DataFrame:
"""获取多标的数据快照"""
...
四、计算层（10个技术指标模块整合）
计算层基于数据层提供的标准化行情数据，计算各类技术分析指标。10个技术指标模块按功能分为趋势指标组、震荡指标组和量能指标组，并配备指标组合引擎。
4.1 趋势指标组
趋势指标组包含MACD（模块3）和EXPMA（模块7），用于识别和跟踪市场趋势方向。
4.1.1 MACD模块（模块3）
学术定义：MACD（Moving Average Convergence Divergence，异同移动平均线）由Gerald Appel于1979年提出，是基于移动平均线的趋势跟踪指标。通过计算短期EMA与长期EMA的差值（DIF线），再对DIF线计算其EMA得到DEA线（信号线），二者的差值绘制为MACD柱状图。
计算公式：
- DIF（快线）= EMA(close, 12) - EMA(close, 26)
- DEA（慢线）= EMA(DIF, 9)
- MACD柱 = 2 × (DIF - DEA)
学术研究用途：
- 趋势方向判断：DIF上穿DEA为金叉（学术术语，非交易信号），下穿为死叉
- 趋势强度判断：MACD柱的绝对值反映趋势强度
- 背离分析：价格创新高/低但MACD未创新高/低，提示趋势可能减弱
class MACDModule:
"""MACD指标计算模块"""
module_name = "macd"
module_version = "1.0.0"
def calculate(
self,
df: pd.DataFrame,
fast_period: int = 12,
slow_period: int = 26,
signal_period: int = 9,
price_field: str = "close"
) -> ModuleResult:
"""
计算MACD指标
参数:
df: 包含价格数据的DataFrame
fast_period: 快线EMA周期（默认12）
slow_period: 慢线EMA周期（默认26）
signal_period: 信号线EMA周期（默认9）
price_field: 价格字段名
返回:
DataFrame包含 dif, dea, macd 三列
"""
close = df[price_field].values
ema_fast = self._calc_ema(close, fast_period)
ema_slow = self._calc_ema(close, slow_period)
dif = ema_fast - ema_slow
dea = self._calc_ema(dif, signal_period)
macd_hist = 2.0 * (dif - dea)
result_df = pd.DataFrame({
"trade_date": df["trade_date"],
"dif": dif,
"dea": dea,
"macd": macd_hist
})
return ModuleResult(
module_name=self.module_name,
data=result_df,
metadata={"fast": fast_period, "slow": slow_period,
"signal": signal_period},
warnings=[], error=None
)
@staticmethod
def _calc_ema(data: np.ndarray, period: int) -> np.ndarray:
"""计算指数移动平均"""
alpha = 2.0 / (period + 1)
ema = np.zeros_like(data, dtype=float)
ema[0] = data[0]
for i in range(1, len(data)):
ema[i] = alpha * data[i] + (1 - alpha) * ema[i - 1]
return ema
4.1.2 EXPMA模块（模块7）
学术定义：EXPMA（Exponential Moving Average，指数移动平均线）通过加权平均方式计算，近期数据权重更大，对价格变化反应更灵敏。
计算公式：
- EXPMA = close × α + 前一日EXPMA × (1 - α)
- 其中 α = 2 / (N + 1)，N为周期参数
class EXPAMAModule:
"""EXPMA指标计算模块"""
module_name = "expma"
module_version = "1.0.0"
def calculate(
self,
df: pd.DataFrame,
short_period: int = 12,
long_period: int = 50,
price_field: str = "close"
) -> ModuleResult:
"""
计算EXPMA指标
返回:
DataFrame包含 expma_short, expma_long 两列
"""
close = df[price_field].values
expma_s = self._calc_ema(close, short_period)
expma_l = self._calc_ema(close, long_period)
result_df = pd.DataFrame({
"trade_date": df["trade_date"],
"expma_short": expma_s,
"expma_long": expma_l,
"expma_diff": expma_s - expma_l
})
return ModuleResult(
module_name=self.module_name,
data=result_df,
metadata={"short": short_period, "long": long_period},
warnings=[], error=None
)
4.2 震荡指标组
震荡指标组包含KDJ（模块2）、WR（模块6）、ARBR（模块8）、DPO（模块9）和VR（模块10），用于判断市场的超买超卖状态和振荡幅度。
4.2.1 KDJ模块（模块2）
学术定义：KDJ（Stochastic Oscillator，随机振荡指标）由George Lane提出，通过比较收盘价与一定周期内的价格区间关系，产生K、D、J三条曲线。
计算公式：
- RSV = (close - LL) / (HH - LL) × 100
- K = SMA(RSV, M1, 1) （即 RSV的M1日移动平均）
- D = SMA(K, M2, 1)
- J = 3K - 2D
其中 HH 和 LL 分别为 N 日内的最高价和最低价。
class KDJModule:
"""KDJ指标计算模块"""
module_name = "kdj"
module_version = "1.0.0"
def calculate(
self,
df: pd.DataFrame,
n: int = 9,
m1: int = 3,
m2: int = 3
) -> ModuleResult:
"""
计算KDJ指标
参数:
n: RSV计算周期（默认9）
m1: K值平滑周期（默认3）
m2: D值平滑周期（默认3）
"""
low_list = df["low"].rolling(window=n, min_periods=1).min()
high_list = df["high"].rolling(window=n, min_periods=1).max()
rsv = (df["close"] - low_list) / (high_list - low_list) * 100.0
k = rsv.ewm(com=m1 - 1, adjust=False).mean()
d = k.ewm(com=m2 - 1, adjust=False).mean()
j = 3.0 * k - 2.0 * d
result_df = pd.DataFrame({
"trade_date": df["trade_date"],
"k": k, "d": d, "j": j
})
return ModuleResult(
module_name=self.module_name,
data=result_df,
metadata={"n": n, "m1": m1, "m2": m2},
warnings=[], error=None
)
4.2.2 WR模块（模块6）
学术定义：WR（Williams %R，威廉指标）由Larry Williams提出，衡量当前价格在N日价格区间中的相对位置。
计算公式：
- WR = (HH - close) / (HH - LL) × (-100)
WR的取值范围为 -100 到 0，学术界通常认为 WR > -20 为超买区域，WR < -80 为超卖区域。
class WRModule:
"""WR指标计算模块"""
module_name = "wr"
module_version = "1.0.0"
def calculate(self, df: pd.DataFrame, period: int = 14) -> ModuleResult:
"""计算WR指标"""
hh = df["high"].rolling(window=period).max()
ll = df["low"].rolling(window=period).min()
wr = (hh - df["close"]) / (hh - ll) * (-100.0)
result_df = pd.DataFrame({
"trade_date": df["trade_date"], "wr": wr
})
return ModuleResult(
module_name=self.module_name,
data=result_df,
metadata={"period": period},
warnings=[], error=None
)
4.2.3 ARBR模块（模块8）
学术定义：ARBR（人气意愿指标）由AR（人气指标）和BR（意愿指标）两部分组成，通过比较开盘价与最高价/最低价的关系来衡量市场买卖人气。
计算公式：
- AR = SUM(high - open, N) / SUM(open - low, N) × 100
- BR = SUM(close - open, N) / SUM(open - close, N) × 100
class ARBRModule:
"""ARBR指标计算模块"""
module_name = "arbr"
module_version = "1.0.0"
def calculate(self, df: pd.DataFrame, period: int = 26) -> ModuleResult:
"""计算ARBR指标"""
ar_up = df["high"] - df["open"]
ar_down = df["open"] - df["low"]
ar = ar_up.rolling(period).sum() / ar_down.rolling(period).sum() * 100
br_up = df["close"] - df["open"]
br_down = df["open"] - df["close"]
br = br_up.clip(lower=0).rolling(period).sum() / \
br_down.clip(lower=0).rolling(period).sum() * 100
result_df = pd.DataFrame({
"trade_date": df["trade_date"], "ar": ar, "br": br
})
return ModuleResult(
module_name=self.module_name,
data=result_df,
metadata={"period": period},
warnings=[], error=None
)
4.2.4 DPO模块（模块9）
学术定义：DPO（Detrended Price Oscillator，去趋势价格振荡器）通过消除价格中的趋势成分，分离出价格的周期性波动成分。
计算公式：
- DPO = close - MA(close, N/2 + 1)
其中位移参数为 N/2 + 1，将移动平均线向后位移以消除趋势。
class DPOModule:
"""DPO指标计算模块"""
module_name = "dpo"
module_version = "1.0.0"
def calculate(self, df: pd.DataFrame, period: int = 20) -> ModuleResult:
"""计算DPO指标"""
shift = period // 2 + 1
ma = df["close"].rolling(window=period).mean().shift(shift)
dpo = df["close"] - ma
result_df = pd.DataFrame({
"trade_date": df["trade_date"], "dpo": dpo
})
return ModuleResult(
module_name=self.module_name,
data=result_df,
metadata={"period": period, "shift": shift},
warnings=[], error=None
)
4.2.5 VR模块（模块10）
学术定义：VR（Volume Ratio，成交量比率）通过比较上涨日成交量与下跌日成交量的比值，衡量市场买卖力量的对比。
计算公式：
- AVS = N日内上涨日成交量之和
- BVS = N日内下跌日成交量之和
- CVS = N日内平盘日成交量之和
- VR = (AVS + CVS/2) / (BVS + CVS/2) × 100
class VRModule:
"""VR指标计算模块"""
module_name = "vr"
module_version = "1.0.0"
def calculate(self, df: pd.DataFrame, period: int = 26) -> ModuleResult:
"""计算VR指标"""
up_vol = df.loc[df["close"] > df["pre_close"], "volume"]
down_vol = df.loc[df["close"] < df["pre_close"], "volume"]
flat_vol = df.loc[df["close"] == df["pre_close"], "volume"]
avs = up_vol.rolling(period, min_periods=1).sum().reindex(df.index).fillna(0)
bvs = down_vol.rolling(period, min_periods=1).sum().reindex(df.index).fillna(0)
cvs = flat_vol.rolling(period, min_periods=1).sum().reindex(df.index).fillna(0)
vr = (avs + cvs / 2) / (bvs + cvs / 2) * 100
result_df = pd.DataFrame({
"trade_date": df["trade_date"], "vr": vr
})
return ModuleResult(
module_name=self.module_name,
data=result_df,
metadata={"period": period},
warnings=[], error=None
)
4.3 量能指标组
量能指标组包含成交量（模块4）和换手率（模块5）两个模块，用于分析市场交易活跃度和量价关系。
4.3.1 成交量模块（模块4）
学术定义：成交量分析是技术分析的基础工具之一，通过研究成交量与价格变动的关系，判断市场趋势的可靠性。
核心分析方法：
- 量价配合：价格上涨伴随放量，趋势可靠性较高
- 量价背离：价格创新高但成交量萎缩，提示趋势动力不足
- 放量突破：价格突破关键位时成交量显著放大
- 缩量整理：价格窄幅波动且成交量萎缩，多处于整理阶段
class VolumeModule:
"""成交量分析模块"""
module_name = "volume"
module_version = "1.0.0"
def calculate(
self,
df: pd.DataFrame,
ma_periods: List[int] = [5, 10, 20]
) -> ModuleResult:
"""
计算成交量指标
返回:
volume: 原始成交量
vol_ma_5/10/20: 成交量移动平均
vol_ratio: 量比（当日成交量 / 过去5日平均成交量）
"""
result = {"trade_date": df["trade_date"], "volume": df["volume"]}
for p in ma_periods:
result[f"vol_ma_{p}"] = df["volume"].rolling(p).mean()
result["vol_ratio"] = df["volume"] / result["vol_ma_5"]
return ModuleResult(
module_name=self.module_name,
data=pd.DataFrame(result),
metadata={"ma_periods": ma_periods},
warnings=[], error=None
)
4.3.2 换手率模块（模块5）
学术定义：换手率（Turnover Rate）指一定时期内股票的成交量占流通股本的百分比，反映股票的流通性和交易活跃程度。
计算公式：
- 换手率 = 成交量 / 流通股本 × 100%
class TurnoverRateModule:
"""换手率分析模块"""
module_name = "turnover_rate"
module_version = "1.0.0"
def calculate(
self,
df: pd.DataFrame,
ma_periods: List[int] = [5, 10, 20]
) -> ModuleResult:
"""
计算换手率指标
返回:
turnover_rate: 原始换手率
tr_ma_5/10/20: 换手率移动平均
tr_rank: 换手率在历史中的分位数
"""
result = {"trade_date": df["trade_date"],
"turnover_rate": df["turnover_rate"]}
for p in ma_periods:
result[f"tr_ma_{p}"] = df["turnover_rate"].rolling(p).mean()
result["tr_rank"] = df["turnover_rate"].rolling(
window=60, min_periods=20
).apply(lambda x: pd.Series(x).rank(pct=True).iloc[-1])
return ModuleResult(
module_name=self.module_name,
data=pd.DataFrame(result),
metadata={"ma_periods": ma_periods},
warnings=[], error=None
)
4.4 指标组合引擎
指标组合引擎负责协调多个技术指标的协同分析，提供共振判断、权重调整和冲突处理机制。
class IndicatorCombinationEngine:
"""指标组合引擎"""
def __init__(self):
self._indicators: Dict[str, IndicatorModule] = {}
self._weights: Dict[str, float] = {}
def register_indicator(self, name: str, module: IndicatorModule,
weight: float = 1.0) -> None:
"""注册指标模块及权重"""
self._indicators[name] = module
self._weights[name] = weight
def check_resonance(self, signals: Dict[str, int]) -> Dict[str, Any]:
"""
共振判断
当多个指标同时发出同方向信号时，称为"共振"。
学术研究表明，多指标共振信号的统计显著性高于单一信号。
参数:
signals: 各指标信号字典 {指标名: 信号方向}
1=看多, -1=看空, 0=中性
返回:
{
"resonance_direction": 共振方向,
"resonance_strength": 共振强度(0-1),
"participating_indicators": 参与共振的指标列表,
"conflicting_indicators": 冲突指标列表
}
"""
weighted_sum = sum(
direction * self._weights.get(name, 1.0)
for name, direction in signals.items()
)
total_weight = sum(self._weights.values())
resonance_strength = abs(weighted_sum) / total_weight
resonance_direction = 1 if weighted_sum > 0 else (
-1 if weighted_sum < 0 else 0
)
participating = [k for k, v in signals.items()
if v == resonance_direction and v != 0]
conflicting = [k for k, v in signals.items()
if v == -resonance_direction and v != 0]
return {
"resonance_direction": resonance_direction,
"resonance_strength": resonance_strength,
"participating_indicators": participating,
"conflicting_indicators": conflicting
}
def dynamic_weight_adjust(self, market_regime: str) -> Dict[str, float]:
"""
动态权重调整
根据市场状态（趋势/震荡/高波动）动态调整各指标权重。
学术研究方法论：趋势市场中趋势指标权重更高，
震荡市场中震荡指标权重更高。
"""
regime_weights = {
"trending": {"macd": 1.5, "expma": 1.5, "kdj": 0.8,
"wr": 0.8, "volume": 1.2},
"ranging": {"macd": 0.8, "expma": 0.8, "kdj": 1.5,
"wr": 1.5, "volume": 1.0},
"volatile": {"macd": 1.0, "expma": 1.0, "kdj": 1.2,
"wr": 1.2, "volume": 1.3},
}
return regime_weights.get(market_regime, {})
五、策略层（学术研究方法论）
策略层运用计算层输出的指标信号，按照学术研究方法论构建策略模板。所有策略描述均以技术分析工具框架和学术研究方法论形式呈现，不构成任何投资建议。
5.1 策略模板库
框架提供四类策略研究模板，每类模板定义了标准化的研究流程和评估方法。
class StrategyTemplate:
"""策略模板基类"""
# 策略类型枚举
STRATEGY_TYPES = {
"trend_following": "趋势跟踪策略",
"mean_reversion": "均值回归策略",
"factor_selection": "因子选股策略",
"event_driven": "事件驱动策略",
}
def __init__(self, name: str, strategy_type: str):
self.name = name
self.strategy_type = strategy_type
self.hypothesis: str = "" # 研究假设
self.applicable_scope: str = "" # 适用范围
self.limitations: List[str] = [] # 局限性说明
def generate_signals(self, data: pd.DataFrame,
params: Dict[str, Any]) -> pd.DataFrame:
"""
生成研究信号
注意：此处生成的信号仅用于学术研究中的历史数据统计分析，
不构成任何买卖建议。
"""
...
def describe(self) -> Dict[str, Any]:
"""输出策略描述（含免责说明）"""
return {
"name": self.name,
"type": self.strategy_type,
"hypothesis": self.hypothesis,
"scope": self.applicable_scope,
"limitations": self.limitations,
"disclaimer": "本策略仅供学术研究参考，不构成投资建议。"
}
四类策略模板说明：
| 策略类型 | 研究假设 | 适用市场状态 | 主要指标 |
|---|---|---|---|
| 趋势跟踪 | 价格趋势具有惯性效应 | 趋势明确的市场 | MACD, EXPMA, 成交量 |
| 均值回归 | 价格偏离均值后趋向回归 | 震荡市场 | KDJ, WR, DPO, ARBR |
| 因子选股 | 特定因子具有预测能力 | 全市场 | 多因子组合 |
| 事件驱动 | 特定事件对价格有短期影响 | 事件密集期 | 新闻, 公告, 政策 |
5.2 事件驱动研究框架
事件驱动研究框架利用资讯数据子系统采集的新闻、公告和政策数据，研究特定事件对证券价格的统计影响。
class EventDrivenResearchFramework:
"""事件驱动研究框架"""
def __init__(self):
self.news_module = MarketNewsModule()
self.announcement_module = CompanyAnnouncementModule()
self.policy_module = PolicyRegulationModule()
def define_event(self, event_config: Dict[str, Any]) -> EventDefinition:
"""
定义研究事件
事件类型:
- 新闻事件: 热点新闻关键词匹配
- 公告事件: 特定类型公告（业绩预告、分红等）
- 政策事件: 宏观政策发布
"""
...
def event_study(
self,
event: EventDefinition,
estimation_window: int = 30, # 估计窗口（事件前30日）
event_window: Tuple[int, int] = (-5, 5), # 事件窗口
post_window: int = 20 # 事后窗口
) -> EventStudyResult:
"""
事件研究法
学术方法论:
1. 定义事件日
2. 计算正常收益率（基于市场模型）
3. 计算异常收益率（AR = 实际收益率 - 正常收益率）
4. 计算累计异常收益率（CAR）
5. 统计显著性检验（t检验）
注意：事件研究结果仅反映历史统计规律，
不代表未来收益，不构成投资建议。
"""
...
def sentiment_analysis(
self,
news_data: pd.DataFrame
) -> pd.DataFrame:
"""
新闻情绪分析
使用NLP技术对新闻文本进行情绪打分，
用于学术研究中的情绪因子构建。
情绪得分范围：-1（极度悲观）到 1（极度乐观）
"""
...
5.3 多因子研究模型
多因子研究模型是量化投资学术研究中的核心方法论之一，通过将多个因子组合构建综合评分模型。
class MultiFactorModel:
"""多因子研究模型"""
def __init__(self):
self.factors: Dict[str, Factor] = {}
def add_factor(self, name: str, factor: Factor) -> None:
"""添加因子"""
self.factors[name] = factor
def build_factor_universe(self, trade_date: date) -> pd.DataFrame:
"""构建因子截面数据"""
...
def factor_preprocessing(
self,
df: pd.DataFrame,
methods: List[str] = ["winsorize", "standardize", "neutralize"]
) -> pd.DataFrame:
"""
因子预处理流水线
1. 去极值（Winsorize）: 将因子值在3倍标准差外的值截断
2. 标准化（Standardize）: Z-Score标准化
3. 中性化（Neutralize）: 去除行业和市值因子的影响
"""
...
def ic_analysis(self, factor_data: pd.DataFrame) -> Dict[str, float]:
"""
IC分析（Information Coefficient）
计算因子值与未来收益率的秩相关系数（Spearman IC），
评估因子的预测能力。
返回:
mean_ic: 平均IC
icir: IC信息比率（mean_ic / std_ic）
positive_ratio: IC为正的比例
"""
...
def factor_combination(
self,
factor_data: pd.DataFrame,
method: str = "equal_weight" # equal_weight/ic_weight/optimal
) -> pd.Series:
"""
因子组合
方法:
- equal_weight: 等权组合
- ic_weight: 按IC加权
- optimal: 最大化IR的优化权重
"""
...
因子分类体系：
| 因子类别 | 包含因子 | 数据来源模块 |
|---|---|---|
| 技术指标因子 | MACD信号、KDJ超买超卖、WR极值、DPO周期、VR量比 | 模块2-10 |
| 情绪因子 | 新闻情绪、投资者互动热度、换手率分位数 | 模块5、11、17 |
| 宏观因子 | 央行政策利率变动、PMI、CPI | 模块16、19 |
| 量价因子 | 量比、换手率变化、波动率、成交额分位数 | 模块4、5、13 |
5.4 信号融合机制
信号融合机制将多个策略模块的输出信号进行综合，生成统一的学术研究信号。
class SignalFusion:
"""信号融合机制"""
def voting_fusion(self, signals: Dict[str, int],
threshold: float = 0.6) -> Dict[str, Any]:
"""
投票法融合
当超过threshold比例的信号方向一致时，输出融合信号。
"""
total = len(signals)
bull_count = sum(1 for v in signals.values() if v > 0)
bear_count = sum(1 for v in signals.values() if v < 0)
if bull_count / total >= threshold:
return {"direction": 1, "confidence": bull_count / total,
"method": "voting"}
elif bear_count / total >= threshold:
return {"direction": -1, "confidence": bear_count / total,
"method": "voting"}
return {"direction": 0, "confidence": 0, "method": "voting"}
def confidence_weighted_fusion(
self,
signals: Dict[str, Tuple[int, float]]
) -> Dict[str, Any]:
"""
置信度加权融合
每个信号附带置信度分数，按置信度加权汇总。
"""
weighted_sum = sum(direction * confidence
for direction, confidence in signals.values())
total_confidence = sum(c for _, c in signals.values())
return {
"direction": 1 if weighted_sum > 0 else (
-1 if weighted_sum < 0 else 0),
"confidence": abs(weighted_sum) / total_confidence,
"method": "confidence_weighted"
}
六、回测层（历史数据验证方法论）
回测层提供完整的历史数据回测验证方法论，用于学术研究中对策略假设的统计检验。回测结果仅反映历史表现，不代表未来收益。
6.1 回测引擎
class BacktestEngine:
"""回测引擎"""
def __init__(self, config: BacktestConfig):
self.config = config
self.trade_calendar = TradeCalendar()
def run(self, strategy: StrategyTemplate,
data: pd.DataFrame) -> BacktestResult:
"""
执行回测
A股特殊规则处理:
- T+1: 当日买入的证券次日方可卖出
- 涨跌停: 涨停价无法买入，跌停价无法卖出
- 停牌: 停牌证券无法交易
- 交易成本: 佣金（万2.5）+ 印花税（卖出千1）+ 过户费
"""
trades = []
portfolio_value = self.config.initial_capital
for i in range(len(data)):
current_date = data.iloc[i]["trade_date"]
# 检查是否为交易日
if not self.trade_calendar.is_trading_day(current_date):
continue
# 生成信号
signal = strategy.generate_signals(
data.iloc[:i+1], self.config.params
)
# 执行交易（考虑T+1、涨跌停等约束）
executed_trades = self._execute_trades(
signal, data.iloc[i], trades, portfolio_value
)
trades.extend(executed_trades)
return BacktestResult(
trades=trades,
portfolio_series=self._calc_portfolio_series(trades, data),
config=self.config
)
def _execute_trades(
self,
signal: pd.DataFrame,
current_bar: pd.Series,
existing_trades: List[Trade],
available_capital: float
) -> List[Trade]:
"""
执行交易（含A股约束处理）
约束检查:
1. T+1检查: 买入的证券当日不可卖出
2. 涨跌停检查: 涨停价不买入，跌停价不卖出
3. 停牌检查: 停牌证券不参与交易
4. 资金检查: 确保有足够资金买入
"""
...
6.2 风控评估
class RiskEvaluator:
"""风控评估模块"""
def evaluate(self, result: BacktestResult) -> RiskReport:
"""
绩效指标计算
指标清单:
- 年化收益率
- 年化波动率
- 夏普比率（Sharpe Ratio）
- 最大回撤（Maximum Drawdown）
- 卡玛比率（Calmar Ratio）
- 索提诺比率（Sortino Ratio）
- 胜率（Win Rate）
- 盈亏比（Profit/Loss Ratio）
- 换手率（Turnover）
- 尾部风险指标（VaR, CVaR）
"""
returns = result.portfolio_series.pct_change().dropna()
annual_return = self._calc_annual_return(returns)
annual_vol = returns.std() * np.sqrt(252)
sharpe = annual_return / annual_vol if annual_vol > 0 else 0
max_drawdown = self._calc_max_drawdown(result.portfolio_series)
var_95 = returns.quantile(0.05)
cvar_95 = returns[returns <= var_95].mean()
return RiskReport(
annual_return=annual_return,
annual_volatility=annual_vol,
sharpe_ratio=sharpe,
max_drawdown=max_drawdown,
var_95=var_95,
cvar_95=cvar_95
)
6.3 稳健性检查
class RobustnessChecker:
"""稳健性检查模块"""
def walk_forward_test(
self,
strategy: StrategyTemplate,
data: pd.DataFrame,
train_size: int = 252,
test_size: int = 63,
step: int = 21
) -> WalkForwardResult:
"""
走动前向测试（Walk-Forward Analysis）
将历史数据分为滚动窗口，每个窗口分为训练期和测试期。
在训练期优化参数，在测试期验证效果。
"""
...
def parameter_perturbation(
self,
strategy: StrategyTemplate,
data: pd.DataFrame,
param_ranges: Dict[str, Tuple[float, float]],
n_iterations: int = 100
) -> ParameterSensitivity:
"""
参数扰动测试
在参数空间内随机采样，评估策略对参数变化的敏感度。
稳健的策略应在参数合理变化范围内保持相对稳定的表现。
"""
...
def cost_pressure_test(
self,
result: BacktestResult,
cost_multipliers: List[float] = [1, 2, 5, 10]
) -> CostSensitivity:
"""
交易成本压力测试
逐步提高交易成本倍数，评估策略对成本的敏感度。
"""
...
def market_regime_test(
self,
strategy: StrategyTemplate,
data: pd.DataFrame
) -> RegimeAnalysis:
"""
市场分层测试
将历史数据按市场状态（牛市/熊市/震荡市）分层，
评估策略在不同市场环境下的表现差异。
"""
...
6.4 归因分析
class AttributionAnalyzer:
"""归因分析模块"""
def industry_attribution(
self,
portfolio: pd.DataFrame,
benchmark: pd.DataFrame,
industry_map: Dict[str, str]
) -> IndustryAttribution:
"""行业归因分析"""
...
def style_attribution(
self,
portfolio: pd.DataFrame,
benchmark: pd.DataFrame,
style_factors: List[str] = ["size", "value", "growth", "momentum"]
) -> StyleAttribution:
"""风格归因分析"""
...
def drawdown_attribution(
self,
portfolio_series: pd.Series,
trades: List[Trade]
) -> DrawdownAttribution:
"""回撤归因分析：识别最大回撤期间的主要亏损来源"""
...
七、执行层（组合管理方法论）
执行层提供组合构建和订单管理的理论方法论，用于学术研究中的组合优化实验。本层内容仅供学习参考，不用于实际交易执行。
7.1 组合构建方法
class PortfolioConstructor:
"""组合构建方法"""
def equal_weight(self, symbols: List[str]) -> Dict[str, float]:
"""等权组合：每个标的分配相同权重"""
n = len(symbols)
return {s: 1.0 / n for s in symbols}
def risk_parity(self, cov_matrix: pd.DataFrame) -> Dict[str, float]:
"""
风险平价组合：每个标的贡献相同的风险
学术原理：使每个资产对组合总风险的贡献相等，
避免单一资产主导组合风险。
"""
...
def volatility_targeting(
self,
expected_returns: pd.Series,
cov_matrix: pd.DataFrame,
target_vol: float = 0.15
) -> Dict[str, float]:
"""
波动率目标组合：通过杠杆调整实现目标波动率
学术原理：根据组合的预期波动率与目标波动率的比值，
调整整体仓位水平。
"""
...
def constrained_optimization(
self,
expected_returns: pd.Series,
cov_matrix: pd.DataFrame,
constraints: Dict[str, Any]
) -> Dict[str, float]:
"""
约束优化组合
约束条件可包括:
- 个股权重上限/下限
- 行业权重上限
- 总仓位限制
- 因子暴露限制
"""
...
7.2 订单管理（理论模型）
class OrderManager:
"""订单管理（理论模型，仅供学术研究）"""
def split_order(
self,
target_quantity: int,
max_single_ratio: float = 0.1,
n_splits: int = 5
) -> List[OrderSlice]:
"""
订单拆分
学术研究用途：研究不同拆单策略对执行成本的影响。
"""
...
def estimate_slippage(
self,
order: OrderSlice,
avg_daily_volume: float
) -> float:
"""
滑点估算模型
学术模型：滑点 ≈ k × (order_size / ADV)^0.5
其中 k 为市场冲击系数，ADV 为日均成交量
"""
...
7.3 实时监控（研究框架）
class MonitoringFramework:
"""实时监控框架（学术研究用途）"""
def portfolio_monitor(self, positions: Dict[str, Position]) -> PortfolioStatus:
"""持仓监控：市值、盈亏、集中度"""
...
def risk_monitor(
self,
positions: Dict[str, Position],
market_data: pd.DataFrame
) -> RiskStatus:
"""风险监控：VaR、希腊字母、相关性"""
...
def news_monitor(
self,
symbols: List[str],
news_data: pd.DataFrame
) -> List[NewsAlert]:
"""新闻监控：关联新闻提醒"""
...
八、A股特殊约束处理
8.1 交易规则
| 规则 | 说明 | 框架处理方式 |
|---|---|---|
| T+1 | 当日买入的证券次日方可卖出 | 回测引擎中维护可卖证券池，标记买入日期 |
| 涨跌停 | 主板10%、创业板/科创板20%、北交所30% | 撮合时检查当前价格是否触及涨跌停 |
| 停牌 | 停牌证券无法交易 | 检查停牌状态，跳过停牌证券的交易信号 |
| 除权除息 | 送股、转增、分红导致价格调整 | 使用复权价格进行计算，记录除权事件 |
class AShareConstraintHandler:
"""A股交易约束处理器"""
def check_t1_constraint(
self,
symbol: str,
action: str,
buy_date: date,
current_date: date
) -> bool:
"""T+1约束检查"""
if action == "sell":
return current_date > buy_date
return True
def check_price_limit(
self,
symbol: str,
price: float,
pre_close: float,
board_type: str # main/gem/star/bse
) -> Tuple[bool, str]:
"""
涨跌停检查
返回:
(是否可交易, 约束类型)
"""
limits = {
"main": 0.10, # 主板 ±10%
"gem": 0.20, # 创业板 ±20%
"star": 0.20, # 科创板 ±20%
"bse": 0.30, # 北交所 ±30%
}
limit_pct = limits.get(board_type, 0.10)
upper = pre_close * (1 + limit_pct)
lower = pre_close * (1 - limit_pct)
if price >= upper:
return False, "涨停"
elif price <= lower:
return False, "跌停"
return True, "正常"
def check_suspension(
self,
symbol: str,
current_date: date
) -> bool:
"""停牌检查"""
...
8.2 特殊标的处理
| 标的类型 | 特殊规则 | 框架处理 |
|---|---|---|
| ST/*ST | 日涨跌限制5%，交易需额外风险提示 | 标记ST状态，调整涨跌停幅度 |
| 注册制新股 | 上市前5日无涨跌限制 | 新股期使用特殊撮合规则 |
| 北交所 | 日涨跌限制30%，最小申报单位100股 | 调整参数适配北交所规则 |
| 可转债 | T+0交易，无涨跌限制（有临停机制） | 单独的交易规则配置 |
8.3 防未来函数
未来函数（Look-Ahead Bias）是量化研究中最常见的错误之一，指在策略计算中使用了当时不可能获得的信息。
class LookAheadBiasGuard:
"""防未来函数守卫"""
def validate_temporal_alignment(
self,
factor_df: pd.DataFrame,
return_df: pd.DataFrame
) -> ValidationResult:
"""
时间对齐验证
确保因子值仅使用截至当日的可得数据计算，
未来收益率使用因子值之后的数据。
"""
...
def check_financial_data_alignment(
self,
report_date: date,
announcement_date: date,
current_date: date
) -> bool:
"""
财务数据对齐检查
上市公司财务报告存在报告期和公告期的差异。
例如，年报的会计截止日为12月31日，
但实际公告日可能在次年3-4月。
研究中必须使用公告日而非报告期作为数据可得日。
"""
return current_date >= announcement_date
def validate_feature_completeness(
self,
feature_df: pd.DataFrame,
as_of_date: date
) -> List[str]:
"""
特征完整性验证
检查截至指定日期，所有使用的特征数据是否已经可得。
"""
...
九、中国法律法规合规要求（重要章节）
本章详细阐述本框架涉及的法律法规合规要求，所有使用者必须严格遵守。
9.1 《中华人民共和国证券法》相关条款
根据2019年修订、2020年3月1日起施行的《中华人民共和国证券法》：
第一百六十条：证券投资咨询机构及其从业人员从事证券服务业务，应当遵守法律、行政法规和中国证监会的规定，勤勉尽责、恪尽职守，按照业务规则和行业规范为投资者提供服务。
第一百六十一条：证券投资咨询机构及其从业人员不得有下列行为：
- 代理委托人从事证券投资
- 与委托人约定分享证券投资收益或者分担证券投资损失
- 买卖本咨询机构提供服务的上市公司股票
- 利用传播媒介或者通过其他方式提供、传播虚假或者误导投资者的信息
第一百六十三条：证券服务机构未勤勉尽责，所制作、出具的文件有虚假记载、误导性陈述或者重大遗漏，给他人造成损失的，应当承担赔偿责任。
9.2 《证券期货投资咨询管理暂行办法》
根据国务院证券委员会1997年发布的《证券期货投资咨询管理暂行办法》：
第二条：本办法所称证券期货投资咨询，是指从事证券期货投资咨询业务的机构及其投资咨询人员，为证券期货投资人或者客户提供证券期货投资分析、预测或者建议等直接或者间接有偿咨询服务的活动。
第三条：从事证券期货投资咨询业务，必须取得中国证监会的业务许可。未经中国证监会许可，任何机构和个人均不得从事本办法第二条所列各种证券期货投资咨询业务。
第二十四条：证券期货投资咨询人员在报刊、电台、电视台或者其他传播媒体上发表投资咨询文章、报告或者意见时，必须注明所在证券期货投资咨询机构的名称和个人真实姓名，并对投资咨询作充分说明。
9.3 证券投资咨询资质要求
在中国境内提供证券投资咨询服务，需满足以下资质要求：
- 机构资质：需取得中国证监会颁发的《经营证券期货业务许可证》，注册资本不低于人民币100万元
- 人员资质：从事证券投资咨询业务的人员需取得证券投资咨询执业资格证书，并在协会注册登记
- 合规管理：建立健全的合规管理制度和内部控制机制
- 信息隔离：建立信息隔离墙制度，防止内幕信息滥用
本框架明确声明：本框架为技术学习工具，不提供任何投资咨询服务。使用者如需投资建议，请咨询持有合法资质的证券投资咨询机构。
9.4 数据合规要求
9.4.1 《中华人民共和国个人信息保护法》（2021年11月1日施行）
- 收集个人信息应当遵循合法、正当、必要原则
- 处理个人信息应当取得个人同意
- 个人信息处理者应当采取必要措施保障个人信息安全
9.4.2 《中华人民共和国数据安全法》（2021年9月1日施行）
- 开展数据处理活动应当依照法律、法规，建立健全数据安全管理制度
- 重要数据的处理应当依法进行安全评估
- 数据处理者应当加强数据安全防护
框架数据合规措施：
- 所有数据仅用于学术研究，不用于商业目的
- 不采集、存储任何个人身份信息
- 数据获取遵守数据源的使用条款和robots协议
- 研究结果不涉及可识别个人或机构的信息
9.5 禁止行为清单
| 序号 | 禁止行为 | 法律依据 |
|---|---|---|
| 1 | 内幕交易 | 《证券法》第五十三条 |
| 2 | 操纵市场 | 《证券法》第五十五条 |
| 3 | 非法荐股 | 《证券法》第一百六十条 |
| 4 | 虚假宣传 | 《广告法》第二十八条 |
| 5 | 未经许可提供投资咨询 | 《证券期货投资咨询管理暂行办法》第三条 |
| 6 | 承诺或暗示投资收益 | 《证券法》第一百六十一条 |
| 7 | 使用未公开信息交易 | 《证券法》第五十三条 |
| 8 | 编造传播虚假证券信息 | 《刑法》第一百八十一条 |
9.6 技术工具的法律边界
合法用途：
- 个人学术研究和学习
- 金融工程方法论研究
- 技术指标计算与分析（不用于实盘交易）
- 历史数据统计分析
- 量化投资教育
非法用途：
- 使用技术工具向不特定对象提供投资建议
- 利用技术工具进行内幕交易或操纵市场
- 将技术分析结果包装为投资建议进行传播
- 未经许可将框架用于商业化的投资咨询服务
9.7 合规自查清单
使用本框架前，请逐项确认以下合规自查清单：
- 我已阅读并理解本文档开头的免责声明
- 我确认本框架仅用于个人学习和学术研究
- 我不会将框架输出的任何结果作为投资决策依据
- 我不会将框架的分析结果传播给他人作为投资建议
- 我不会将框架用于任何商业化的投资咨询活动
- 我理解量化分析存在模型风险，历史表现不代表未来收益
- 我已了解中国证券市场相关法律法规
- 我的数据获取和使用方式符合相关法律法规要求
- 我承诺在分享研究成果时附带完整的免责声明
- 我理解违反相关法律法规可能承担的法律责任
十、完整框架代码实现
以下提供完整的框架主类实现，整合所有20个模块。
"""
A股量化投研一体化框架 - 完整实现
版本: 1.0.0
用途: 技术学习和学术研究
免责: 不构成任何投资建议
"""
import numpy as np
import pandas as pd
from abc import ABC, abstractmethod
from typing import Dict, List, Optional, Any, Tuple, Protocol
from dataclasses import dataclass, field
from datetime import datetime, date
from enum import Enum
import logging
import warnings
# ============================================================
# 基础数据结构
# ============================================================
@dataclass
class ModuleResult:
"""模块统一输出格式"""
module_name: str
data: pd.DataFrame
metadata: Dict[str, Any] = field(default_factory=dict)
warnings: List[str] = field(default_factory=list)
error: Optional[str] = None
@dataclass
class BacktestConfig:
"""回测配置"""
initial_capital: float = 1_000_000.0
commission_rate: float = 0.00025 # 佣金万2.5
stamp_tax_rate: float = 0.001 # 印花税千1（卖出）
slippage: float = 0.001 # 滑点千1
benchmark: str = "000300.SH" # 基准指数
@dataclass
class TradeRecord:
"""交易记录"""
symbol: str
direction: str # buy/sell
price: float
quantity: int
trade_date: date
commission: float
stamp_tax: float = 0.0
slippage_cost: float = 0.0
@dataclass
class BacktestResult:
"""回测结果"""
trades: List[TradeRecord]
portfolio_series: pd.Series
benchmark_series: pd.Series
config: BacktestConfig
metrics: Dict[str, float] = field(default_factory=dict)
# ============================================================
# 模块1: 股市时间校准
# ============================================================
class TradeCalendar:
"""交易日历模块"""
module_name = "trade_calendar"
def __init__(self):
self._holidays: set = set()
self._trading_days: List[date] = []
def load_calendar(self, year: int) -> None:
"""加载指定年份交易日历"""
# 实际实现需从数据源获取
...
def is_trading_day(self, d: date) -> bool:
"""判断是否为交易日"""
if d.weekday() >= 5:
return False
return d not in self._holidays
def get_previous_trading_day(self, d: date) -> date:
"""获取前一个交易日"""
...
def get_next_trading_day(self, d: date) -> date:
"""获取后一个交易日"""
...
def get_trading_days_between(
self, start: date, end: date
) -> List[date]:
"""获取两个日期之间的所有交易日"""
...
# ============================================================
# 主框架类
# ============================================================
class AShareQuantFramework:
"""
A股量化投研一体化框架主类
整合20个模块的五层架构框架，仅供技术学习和学术研究。
"""
DISCLAIMER = (
"免责声明：本框架仅供技术学习和学术研究参考，"
"不构成任何投资建议。股市有风险，投资需谨慎。"
)
def __init__(self, config: Optional[Dict[str, Any]] = None):
self.config = config or {}
self.logger = self._setup_logger()
# 五层架构组件
self.data_layer = DataManager()
self.calc_layer = CalculationLayer()
self.strategy_layer = StrategyLayer()
self.backtest_layer = BacktestLayer()
self.execution_layer = ExecutionLayer()
# 交易日历
self.calendar = TradeCalendar()
# A股约束处理器
self.constraint_handler = AShareConstraintHandler()
self.bias_guard = LookAheadBiasGuard()
self.logger.info(
f"AShareQuantFramework 初始化完成 - {self.DISCLAIMER}"
)
def _setup_logger(self) -> logging.Logger:
"""配置日志系统"""
logger = logging.getLogger("AShareQuant")
logger.setLevel(logging.INFO)
handler = logging.StreamHandler()
handler.setFormatter(logging.Formatter(
'%(asctime)s - %(name)s - %(levelname)s - %(message)s'
))
logger.addHandler(handler)
return logger
# ---- 模块注册与初始化 ----
def register_all_modules(self) -> None:
"""注册全部20个模块"""
self.logger.info("开始注册20个模块...")
# 数据层模块（10个）
self.data_layer.register("historical_kline", HistoricalKLineModule())
self.data_layer.register("index_kline", IndexKLineModule())
self.data_layer.register("call_auction", CallAuctionModule())
self.data_layer.register("market_news", MarketNewsModule())
self.data_layer.register("company_announcement",
CompanyAnnouncementModule())
self.data_layer.register("policy_regulation",
PolicyRegulationModule())
self.data_layer.register("secretary_interaction",
SecretaryInteractionModule())
self.data_layer.register("broker_research", BrokerResearchModule())
self.data_layer.register("central_bank_policy",
CentralBankPolicyModule())
self.data_layer.register("us_market_news", USMarketNewsModule())
# 计算层模块（10个）
self.calc_layer.register("macd", MACDModule())
self.calc_layer.register("kdj", KDJModule())
self.calc_layer.register("volume", VolumeModule())
self.calc_layer.register("turnover_rate", TurnoverRateModule())
self.calc_layer.register("wr", WRModule())
self.calc_layer.register("expma", EXPAMAModule())
self.calc_layer.register("arbr", ARBRModule())
self.calc_layer.register("dpo", DPOModule())
self.calc_layer.register("vr", VRModule())
self.logger.info("20个模块注册完成")
# ---- 数据获取流水线 ----
def fetch_data_pipeline(
self,
symbols: List[str],
start_date: date,
end_date: date,
data_types: List[str] = None
) -> Dict[str, pd.DataFrame]:
"""
数据获取流水线
参数:
symbols: 证券代码列表
start_date: 开始日期
end_date: 结束日期
data_types: 需要获取的数据类型列表
["kline", "index", "news", "announcement",
"policy", "research"]
返回:
各类数据的字典
"""
data_types = data_types or ["kline", "index"]
results = {}
if "kline" in data_types:
for symbol in symbols:
result = self.data_layer.get_module(
"historical_kline"
).fetch_daily(symbol, start_date, end_date)
if result.error is None:
results[f"kline_{symbol}"] = result.data
if "index" in data_types:
for idx in ["000001.SH", "399001.SZ", "399006.SZ"]:
result = self.data_layer.get_module(
"index_kline"
).fetch_index_daily(idx, start_date, end_date)
if result.error is None:
results[f"index_{idx}"] = result.data
if "news" in data_types:
result = self.data_layer.get_module(
"market_news"
).fetch_hot_news(end_date)
if result.error is None:
results["news"] = result.data
return results
# ---- 指标计算流水线 ----
def calculate_indicators_pipeline(
self,
kline_data: pd.DataFrame,
indicators: List[str] = None
) -> Dict[str, pd.DataFrame]:
"""
指标计算流水线
参数:
kline_data: K线数据
indicators: 需要计算的指标列表
["macd", "kdj", "wr", "expma", "arbr",
"dpo", "vr", "volume", "turnover_rate"]
返回:
各指标计算结果的字典
"""
indicators = indicators or [
"macd", "kdj", "wr", "expma", "arbr",
"dpo", "vr", "volume", "turnover_rate"
]
results = {}
for ind_name in indicators:
module = self.calc_layer.get_module(ind_name)
if module is not None:
result = module.calculate(kline_data)
if result.error is None:
results[ind_name] = result.data
self.logger.info(
f"指标 {ind_name} 计算完成"
)
return results
# ---- 策略研究信号生成 ----
def generate_research_signals(
self,
indicator_results: Dict[str, pd.DataFrame],
strategy_type: str = "multi_factor"
) -> pd.DataFrame:
"""
策略研究信号生成
注意：生成的信号仅用于学术研究中的历史统计分析，
不构成任何买卖建议。
参数:
indicator_results: 指标计算结果
strategy_type: 策略类型
返回:
信号DataFrame
"""
self.logger.warning(
"研究信号仅供学术分析，不构成投资建议"
)
signals = {}
for name, df in indicator_results.items():
if name == "macd":
signals["macd"] = np.where(
df["dif"] > df["dea"], 1,
np.where(df["dif"] < df["dea"], -1, 0)
)
elif name == "kdj":
signals["kdj"] = np.where(
df["k"] > df["d"], 1,
np.where(df["k"] < df["d"], -1, 0)
)
elif name == "wr":
signals["wr"] = np.where(
df["wr"] > -20, -1,
np.where(df["wr"] < -80, 1, 0)
)
signal_df = pd.DataFrame(signals)
signal_df["trade_date"] = list(
indicator_results.values()
)[0]["trade_date"].values[:len(signal_df)]
return signal_df
# ---- 回测执行 ----
def run_backtest(
self,
signals: pd.DataFrame,
kline_data: pd.DataFrame,
config: Optional[BacktestConfig] = None
) -> BacktestResult:
"""
执行回测
注意：回测结果仅反映历史统计规律，不代表未来收益。
"""
config = config or BacktestConfig()
self.logger.info("开始执行回测...")
engine = BacktestEngine(config)
result = engine.run(
SimpleStrategy(), kline_data
)
# 风控评估
evaluator = RiskEvaluator()
risk_report = evaluator.evaluate(result)
result.metrics = {
"annual_return": risk_report.annual_return,
"sharpe_ratio": risk_report.sharpe_ratio,
"max_drawdown": risk_report.max_drawdown,
}
self.logger.info(
f"回测完成 - 年化收益: {risk_report.annual_return:.2%}, "
f"夏普比率: {risk_report.sharpe_ratio:.2f}, "
f"最大回撤: {risk_report.max_drawdown:.2%}"
)
self.logger.warning(
"以上回测结果仅供学术研究，不构成投资建议"
)
return result
# ---- 结果输出与可视化 ----
def visualize_results(
self,
result: BacktestResult,
output_dir: str = "./output"
) -> None:
"""
结果可视化
输出图表:
- 净值曲线对比图
- 回撤曲线图
- 月度收益热力图
- 持仓分布饼图
"""
import matplotlib.pyplot as plt
import matplotlib.dates as mdates
fig, axes = plt.subplots(2, 2, figsize=(14, 10))
# 净值曲线
ax1 = axes[0, 0]
ax1.plot(result.portfolio_series.index,
result.portfolio_series.values,
label="策略净值", color="blue")
ax1.plot(result.benchmark_series.index,
result.benchmark_series.values,
label="基准净值", color="gray", linestyle="--")
ax1.set_title("净值曲线对比（仅供学术研究）")
ax1.legend()
# 回撤曲线
ax2 = axes[0, 1]
drawdown = result.portfolio_series / \
result.portfolio_series.cummax() - 1
ax2.fill_between(drawdown.index, drawdown.values, 0,
color="red", alpha=0.3)
ax2.set_title("回撤曲线")
plt.tight_layout()
plt.savefig(f"{output_dir}/backtest_result.png", dpi=150)
self.logger.info(f"可视化结果已保存至 {output_dir}/")
# ============================================================
# 辅助类
# ============================================================
class DataManager:
"""数据管理层"""
def __init__(self):
self._modules: Dict[str, Any] = {}
def register(self, name: str, module: Any) -> None:
self._modules[name] = module
def get_module(self, name: str) -> Any:
return self._modules.get(name)
class CalculationLayer:
"""计算层"""
def __init__(self):
self._modules: Dict[str, Any] = {}
def register(self, name: str, module: Any) -> None:
self._modules[name] = module
def get_module(self, name: str) -> Any:
return self._modules.get(name)
class StrategyLayer:
"""策略层"""
pass
class BacktestLayer:
"""回测层"""
pass
class ExecutionLayer:
"""执行层"""
pass
class SimpleStrategy:
"""简单策略模板（学术演示用）"""
pass
十一、学术研究案例
重要提示：以下所有案例仅供学术研究参考，不构成任何投资建议。案例中的策略表现基于历史数据回测，不代表未来收益。历史业绩不代表投资收益承诺。
11.1 多因子选股研究
研究目的：探索技术指标因子在A股市场中的选股效果。
研究假设：H0 - 技术指标因子对未来N日收益率无预测能力；H1 - 技术指标因子对未来N日收益率有预测能力。
研究方法：
- 因子构建：使用MACD信号、KDJ超买超卖指标、换手率分位数、成交量比率构建因子
- 因子预处理：去极值（3σ）、标准化、行业中性化
- 分层回测：按因子值将股票分为5组，比较各组平均收益率
- IC分析：计算因子值与未来收益率的Spearman秩相关系数
- 多因子组合：使用等权法组合多个因子
学术研究代码示例：
def multi_factor_research_case():
"""
多因子选股研究案例
免责声明：本案例仅供学术研究参考，不构成投资建议。
"""
print("=" * 60)
print("多因子选股研究 - 仅供学术参考")
print("=" * 60)
# 步骤1: 获取数据
framework = AShareQuantFramework()
framework.register_all_modules()
data = framework.fetch_data_pipeline(
symbols=["sample_pool"],
start_date=date(2020, 1, 1),
end_date=date(2024, 12, 31)
)
# 步骤2: 计算因子
indicators = framework.calculate_indicators_pipeline(
data["kline_sample_pool"]
)
# 步骤3: 因子预处理
# 去极值 -> 标准化 -> 行业中性化
...
# 步骤4: 分层回测
# 按综合因子得分分5组，比较各组收益
...
# 步骤5: 输出结果（仅供学术参考）
print("研究结果仅供学术研究参考，不构成投资建议")
print("历史回测表现不代表未来收益")
11.2 事件驱动研究
研究目的：研究上市公司业绩预告事件对股价的短期影响。
研究方法：事件研究法（Event Study）
研究步骤：
- 定义事件：业绩预告公告发布日为事件日（t=0）
- 估计窗口：事件前30个交易日（t=-30至t=-1）
- 事件窗口：事件前后各5个交易日（t=-5至t=5）
- 计算正常收益率：使用市场模型估计
- 计算异常收益率：AR = 实际收益率 - 正常收益率
- 计算累计异常收益率：CAR = ΣAR
- 统计检验：t检验判断CAR是否显著异于零
def event_driven_research_case():
"""
事件驱动研究案例
免责声明：本案例仅供学术研究参考，不构成投资建议。
"""
print("=" * 60)
print("事件驱动研究 - 仅供学术参考")
print("=" * 60)
# 事件研究法实现
# 1. 收集业绩预告公告数据
# 2. 计算异常收益率
# 3. 统计检验
print("事件研究结果仅供学术研究参考")
11.3 大盘择时研究
研究目的：研究多指标共振信号对大盘指数走势的预测能力。
研究方法：使用MACD、KDJ、VR、ARBR四个指标构建择时信号，通过回测验证信号的历史表现。
def market_timing_research_case():
"""
大盘择时研究案例
免责声明：本案例仅供学术研究参考，不构成投资建议。
"""
print("=" * 60)
print("大盘择时研究 - 仅供学术参考")
print("=" * 60)
framework = AShareQuantFramework()
framework.register_all_modules()
# 获取指数数据
data = framework.fetch_data_pipeline(
symbols=["000300.SH"],
start_date=date(2018, 1, 1),
end_date=date(2024, 12, 31),
data_types=["kline", "index"]
)
# 计算指标
indicators = framework.calculate_indicators_pipeline(
data["index_000300.SH"]
)
# 生成择时信号（学术研究用途）
signals = framework.generate_research_signals(indicators)
# 回测验证
result = framework.run_backtest(signals, data["index_000300.SH"])
print("择时研究结果仅供学术研究参考，不构成投资建议")
十二、部署与运维
12.1 系统部署
环境要求：
| 组件 | 最低要求 | 推荐配置 |
|---|---|---|
| Python | 3.8+ | 3.10+ |
| 内存 | 4GB | 16GB+ |
| 磁盘 | 20GB | 100GB+ SSD |
| 操作系统 | Linux/Windows/macOS | Ubuntu 22.04 LTS |
依赖安装：
# 创建虚拟环境
python -m venv quant_env
source quant_env/bin/activate # Linux/Mac
# quant_env\Scripts\activate # Windows
# 安装依赖
pip install numpy pandas matplotlib
pip install scikit-learn # 可选，用于机器学习因子
pip install sqlalchemy # 可选，用于数据存储
# 验证安装
python -c "import ashare_quant; print(ashare_quant.__version__)"
12.2 定时任务
from apscheduler.schedulers.background import BackgroundScheduler
def setup_scheduled_tasks(framework: AShareQuantFramework):
"""
设置定时任务
注意：定时任务仅用于数据更新和学术研究，
不用于自动交易。
"""
scheduler = BackgroundScheduler()
# 每个交易日收盘后更新数据（15:30）
scheduler.add_job(
framework.data_layer.incremental_update,
trigger="cron",
hour=15, minute=30,
day_of_week="mon-fri",
args=["all_modules"]
)
# 每周一生成周度研究报告
scheduler.add_job(
generate_weekly_report,
trigger="cron",
hour=9, minute=0,
day_of_week="mon"
)
scheduler.start()
12.3 监控告警
class SystemMonitor:
"""系统监控模块"""
def check_data_freshness(self) -> List[str]:
"""检查数据时效性"""
alerts = []
for name, module in self.data_layer._modules.items():
last_update = module.get_last_update_time()
if last_update and last_update.date() < date.today():
alerts.append(
f"数据模块 {name} 最后更新于 {last_update}"
)
return alerts
def check_system_health(self) -> Dict[str, Any]:
"""系统健康检查"""
return {
"status": "healthy",
"modules_loaded": len(self.data_layer._modules),
"memory_usage": self._get_memory_usage(),
"disk_usage": self._get_disk_usage(),
}
12.4 日志管理
def setup_logging(config: Dict[str, Any]) -> None:
"""
配置日志系统
日志级别:
- DEBUG: 调试信息
- INFO: 常规运行信息
- WARNING: 警告信息（如数据异常）
- ERROR: 错误信息（如模块执行失败）
日志文件按日期轮转，保留30天。
"""
from logging.handlers import TimedRotatingFileHandler
logger = logging.getLogger("AShareQuant")
logger.setLevel(logging.INFO)
# 文件处理器（按天轮转）
file_handler = TimedRotatingFileHandler(
filename="logs/ashare_quant.log",
when="midnight",
backupCount=30,
encoding="utf-8"
)
file_handler.setFormatter(logging.Formatter(
'%(asctime)s - %(name)s - %(levelname)s - %(message)s'
))
logger.addHandler(file_handler)
十三、免责声明与风险提示
13.1 完整法律免责声明
免责声明：
-
本文档及其中描述的"A股量化投研一体化框架"（以下简称"本框架"）仅供技术学习、学术研究和教育用途，不构成任何形式的投资建议、投资咨询、证券交易指导或投资收益承诺。
-
本框架中涉及的所有策略、模型、指标、代码和案例均为技术演示和学术研究用途，不保证其准确性、完整性、时效性和适用性。任何基于本框架内容进行的投资决策，均由投资者自行承担全部风险和责任。
-
根据《中华人民共和国证券法》及《证券期货投资咨询管理暂行办法》的规定，提供投资建议需持有中国证监会颁发的证券投资咨询执业资格证书。本框架的开发者和维护者未持有上述资格证书，因此不提供任何投资咨询服务。
-
本框架不推荐任何具体股票、基金或其他金融产品的买入、卖出或持有操作。文档中出现的任何证券代码仅作为技术演示的数据标识，不构成对该证券的投资建议。
-
股市有风险，投资需谨慎。投资者应根据自身风险承受能力、投资经验和财务状况，独立做出投资决策，并在必要时咨询持有合法资质的专业投资顾问。
-
本框架使用的数据来源于公开信息渠道，不对数据的准确性、完整性和及时性做出任何保证。数据可能存在延迟、错误或遗漏。
-
本免责声明适用于本框架的所有版本和所有组成部分，包括但不限于文档、代码、配置文件和输出结果。
13.2 风险分类说明
| 风险类型 | 说明 |
|---|---|
| 技术风险 | 软件可能存在缺陷或错误，导致计算结果不准确；数据接口可能中断或变更 |
| 数据风险 | 历史数据可能存在错误、缺失或偏差；数据源可能中断；数据延迟可能导致分析结果失真 |
| 模型风险 | 量化模型基于历史数据构建，历史规律不一定在未来重复；模型参数可能过拟合；极端市场环境下模型可能失效 |
| 法律风险 | 不当使用本框架可能违反相关法律法规，包括但不限于非法荐股、内幕交易、操纵市场等 |
| 市场风险 | 证券市场存在系统性风险，包括政策风险、经济周期风险、流动性风险等，可能导致投资损失 |
13.3 投资者教育建议
- 学习基础知识：在接触量化投资之前，建议先学习证券市场基础知识、金融学和统计学基础
- 理解模型局限：任何量化模型都有局限性，不存在"万能"的交易策略
- 重视风险管理：投资的核心是风险管理而非收益最大化
- 警惕高收益承诺：任何承诺高额稳定收益的产品或服务都值得警惕
- 选择合法渠道：投资咨询请选择持有中国证监会颁发资质的合法机构
- 独立思考判断：不盲目跟风，独立做出投资决策
- 持续学习更新：金融市场不断变化，需要持续学习和更新知识
最终提醒：本框架的全部内容——包括文档、代码、策略、案例和任何输出结果——均仅供技术学习和学术研究参考。请严格遵守中国法律法规，理性对待量化投资，切勿将学习工具误用于实际投资决策。如需投资建议，请咨询持有合法资质的证券投资咨询机构。
文档版本：7.1.09
最后更新：2026年6月
用途：技术学习和学术研究
再次声明：本文档全部内容仅供技术学习和学术研究参考，不构成任何投资建议。股市有风险，投资需谨慎。