DEV Community: Henry Lin

千万用户股票量化平台设计

Henry Lin — Fri, 10 Jul 2026 06:06:37 +0000

千万用户股票量化平台设计

日期： 2026-07-10

状态： 已确认总体路线

架构路线： 云原生 Lakehouse + 流批一体 + 异步计算平面

1. 目标与容量基线

平台面向量化研究、因子生产、历史回测和 AI Research，不承担交易所级撮合或微秒级高频交易。

指标	目标值
注册用户	1,000 万
日活用户	100 万
峰值在线用户	10 万
API 峰值	20 万 QPS
标的范围	A 股约 6,000 个标的
行情粒度	分钟级，架构可扩展至 Tick
日回测任务	约 5 万个
核心查询可用性	99.95%
缓存命中 API 延迟	p95 < 100 ms
普通 API 延迟	p95 < 200 ms
分钟行情时效	分钟结束后 10 秒内可查询

2. 设计原则

Iceberg 数据湖是历史事实源，ClickHouse、Redis 和 OpenSearch 均为可重建索引。
在线请求与因子、回测、AI 等重计算任务运行在隔离的资源池。
所有研究结果绑定代码版本、参数、数据快照和成本模型，保证可复现。
公共行情尽量由 CDN 和 Redis 承载，防止峰值流量穿透事务数据库。
Agent 只能调用受控工具，不直接访问生产数据库或执行任意代码。
基础设施按阶段扩容，不按 1,000 万用户一次性采购。

3. 总体架构

flowchart TB
    subgraph Sources["数据源"]
        EX["交易所 / 行情供应商"]
        FIN["财务 / 公告 / 新闻"]
        ALT["宏观 / 舆情 / 另类数据"]
    end

    subgraph Ingestion["数据采集层"]
        COL["行情采集器集群"]
        CDC["CDC / 批量连接器"]
        MQ["Kafka 事件总线"]
        DQ["Schema Registry + 数据质量"]
    end

    subgraph Lake["数据湖与查询层"]
        OBJ["对象存储 S3 / OSS"]
        ICE["Apache Iceberg Lakehouse"]
        CAT["数据目录、血缘与权限"]
        CH["ClickHouse 实时分析"]
        OS["OpenSearch 文本与向量检索"]
    end

    subgraph Compute["量化计算平面"]
        FLINK["Flink 实时清洗与分钟因子"]
        SPARK["Spark 批量清洗与历史因子"]
        FACTOR["因子注册中心"]
        RAY["Ray 回测与参数搜索"]
    end

    subgraph AI["AI Research 与 Agent"]
        RAG["Research RAG"]
        LLM["LLM Gateway"]
        AGENT["Agent Orchestrator"]
        TOOLS["查询 / 因子 / 回测 / 报告工具"]
        GUARD["权限 / 沙箱 / 审计 / 成本限额"]
    end

    subgraph Serving["在线服务平面"]
        GW["CDN + WAF + API Gateway"]
        BFF["REST / GraphQL / WebSocket"]
        SVC["用户 / 策略 / 组合 / 任务服务"]
        CACHE["Redis Cluster"]
        DB["分片 PostgreSQL"]
        JOB["任务队列与调度器"]
    end

    subgraph Platform["Kubernetes 平台"]
        K8S["多可用区 Kubernetes"]
        OBS["Metrics / Logs / Traces"]
        IAM["IAM / KMS / Secrets"]
        CICD["GitOps / CI/CD"]
    end

    EX --> COL
    FIN --> CDC
    ALT --> CDC
    COL --> MQ
    CDC --> MQ
    MQ --> DQ
    MQ --> FLINK
    DQ --> OBJ
    OBJ --- ICE
    ICE --- CAT
    FLINK --> ICE
    FLINK --> CH
    SPARK <--> ICE
    SPARK --> FACTOR
    FACTOR --> ICE
    GW --> BFF
    BFF --> SVC
    BFF --> CACHE
    SVC --> DB
    SVC --> JOB
    JOB --> RAY
    RAY <--> ICE
    RAY --> OBJ
    ICE --> RAG
    OS --> RAG
    RAG --> LLM
    LLM --> AGENT
    AGENT --> GUARD
    GUARD --> TOOLS
    TOOLS --> CH
    TOOLS --> RAY
    TOOLS --> FACTOR
    K8S -.承载.-> Ingestion
    K8S -.承载.-> Compute
    K8S -.承载.-> AI
    K8S -.承载.-> Serving
    OBS -.监控.-> K8S
    IAM -.安全控制.-> K8S
    CICD -.部署.-> K8S

4. 技术选型

领域	推荐技术	说明
API 入口	CDN、WAF、Envoy Gateway 或 Kong	鉴权、限流、灰度和 WebSocket
在线服务	Go/Java；量化和 AI 使用 Python	隔离在线性能与研究生态
消息总线	Kafka、Schema Registry	顺序消费、解耦和行情回放
实时计算	Flink、Flink Kubernetes Operator	有状态计算和一致性检查点
数据湖	S3/OSS、Iceberg、Parquet	快照、时间旅行和低成本存储
批量计算	Spark on Kubernetes	清洗、历史因子和数据回填
因子计算	Spark、Flink、Python、Numba	支持日频和分钟级因子
回测	Ray、Kubernetes Jobs	参数搜索和弹性资源调度
实时查询	ClickHouse	K 线、因子截面和回测分析
缓存	Redis Cluster	热点行情、会话和限流计数
事务数据	PostgreSQL 16，多分片多副本	用户、权限、策略和任务元数据
搜索与 RAG	OpenSearch	公告、研报、新闻和混合检索
Agent 工作流	Temporal、自研 Orchestrator	持久状态、重试和人工审批
数据编排	Airflow 或 Dagster	日终任务和依赖调度
可观测性	Prometheus、Grafana、Loki、Tempo	指标、日志和链路追踪
交付	Kubernetes、Argo CD、Terraform	GitOps 和环境一致性

优先选用云厂商的托管 Kafka、PostgreSQL、Redis、对象存储和 KMS。Flink、Spark、Ray 与业务服务保持 Kubernetes 可移植性，降低跨云切换成本。

5. 数据模型与数据流

数据分为四层：

raw：供应商原始数据，只追加，不覆盖。
standardized：完成时区、交易日历、代码映射、去重、复权和质量标记。
feature：因子值、标签、横截面标准化结果及版本元数据。
serving：写入 ClickHouse、Redis 和 OpenSearch 的查询优化副本。

sequenceDiagram
    participant V as 数据供应商
    participant K as Kafka
    participant F as Flink
    participant L as Iceberg
    participant C as ClickHouse
    participant R as 因子/回测集群
    participant A as AI Agent
    participant G as API Gateway
    participant U as 用户

    V->>K: 行情、财务、公告、新闻
    K->>F: 实时校验、去重、复权、聚合
    F->>L: 原始层和标准层
    F->>C: 实时行情与分钟因子
    L->>R: 固定版本的数据快照
    R->>L: 因子值、回测结果和制品
    U->>G: 查询、研究或提交回测
    G->>C: 查询热点数据
    G->>R: 异步提交计算任务
    A->>L: 检索可信数据
    A->>R: 调用受控因子和回测工具
    A-->>U: 带引用的结论、报告和任务状态

Kafka 事件和 Iceberg 表都携带 source、schema_version、event_time、ingest_time、trading_date 和 quality_flags。因子结果额外携带 factor_version、code_commit、data_snapshot_id 和 parameters_hash。

6. 扩容方案

6.1 服务与 Kubernetes

在线、流处理、批量回测、AI 使用独立节点池和资源配额。
API 按 CPU、QPS 和延迟使用 HPA；消费者按 Kafka Lag 使用 KEDA。
回测节点池可缩容到零，普通任务优先使用 Spot/抢占式实例。
PodDisruptionBudget、TopologySpreadConstraints 和反亲和保证跨可用区部署。

6.2 数据系统

Kafka 初始设置 192 至 384 个分区，按市场和证券代码哈希分配，Broker 跨三个可用区。
Flink 并行度跟随 Kafka 分区和消费延迟调整，扩缩容通过 Savepoint 完成。
Iceberg 按交易日期分区、证券代码分桶，周期性合并小文件和清理过期快照。
ClickHouse 使用 4 至 8 个分片、每分片 2 个副本，近期数据放本地 SSD，历史数据放对象存储。
PostgreSQL 按 user_id 分片，读副本承担非强一致查询，避免跨分片事务。
CDN 与 Redis 的公共行情缓存命中率目标不低于 80%。

6.3 租户和任务治理

每个租户设置 API、并发回测、CPU 小时、存储和 LLM Token 配额。
回测队列区分交互式、批量和内部任务，采用加权公平调度。
对参数组合数、最大回测区间、单任务运行时间和输出大小设置硬限制。

7. 容灾方案

数据等级	方案	RPO	RTO
用户、策略、权限	跨可用区同步复制，异地 WAL 归档	5 分钟	30 分钟
Kafka 行情流	三副本，MirrorMaker 跨地域复制	5 分钟	30 分钟
Iceberg 数据湖	版本控制和跨地域对象复制	1 小时	2 小时
ClickHouse/OpenSearch	双副本，可从数据湖重建	15 分钟	2 小时
回测任务	Temporal 持久状态，失败后重调度	15 分钟	4 小时

主地域部署三个可用区，异地保持 20% 至 30% 温备容量。Terraform 状态、GitOps 仓库、镜像仓库和密钥备份必须独立于主地域。每月执行备份恢复，每季度执行地域切换、Kafka 重放和节点故障演练。

8. AI Research 与 Agent

Research RAG 使用关键词与向量混合检索，答案必须带数据日期、来源和文档引用。LLM Gateway 负责模型路由、缓存、脱敏、Token 预算、超时和降级。

Agent 工具采用显式 JSON Schema，限定可查询数据集和最大结果量。外部研报作为不可信输入处理。高成本回测、大规模导出以及未来可能增加的实盘操作必须经过人工审批。所有工具调用写入不可篡改审计日志。

9. 失败处理与质量保障

采集失败进入重试主题和死信主题；恢复后按事件时间重放。
数据质量门禁覆盖完整性、唯一性、价格边界、交易日历和跨源对账。
因子发布前检查未来函数、幸存者偏差、缺失率、极值和稳定性。
回测采用确定性随机种子，并验证手续费、滑点、停牌、涨跌停和公司行动。
下游索引写入采用幂等键，失败后从 Iceberg 快照重建。
发布前执行容量压测、故障注入和完整地域恢复演练。

10. 成本估算

以下为满负载月度人民币估算，不含税，实际成本受云厂商折扣、平均 QPS、回测 CPU 小时和模型调用量影响。

项目	月成本（万元）
在线 Kubernetes、Gateway、CDN	45-90
Kafka、Flink	25-55
对象存储、Iceberg、目录服务	10-30
ClickHouse	25-60
PostgreSQL、Redis、OpenSearch	25-60
Spark、Ray 弹性计算	50-180
网络、监控、安全	25-70
异地容灾	40-100
LLM 推理或 API	30-200
合计	275-845

行情、公告和研报授权通常另需 20 至 200 万元以上/月。MVP 阶段预计 20 至 50 万元/月，增长阶段预计 80 至 180 万元/月。预留实例、Spot 回测节点、冷热分层和模型路由预计可降低 30% 至 50% 的可优化成本。

11. 分阶段交付

基础期：日线/分钟线数据湖、批量因子、单策略回测、基础 API 和统一身份。
增长期：Flink 实时链路、ClickHouse、分布式 Ray 回测、租户配额和完整可观测性。
智能期：Research RAG、受控 Agent、报告生成和模型成本治理。
规模期：数据库分片、跨地域温备、20 万 QPS 压测和自动扩缩容优化。

每一阶段均以可观测的容量阈值触发下一阶段，而不是仅以注册用户数判断扩容。

Quant Agent 完整系统规格

Henry Lin — Fri, 10 Jul 2026 06:04:46 +0000

Quant Agent 完整系统规格

文档状态： Approved Design

版本： 1.0

日期： 2026-07-10

目标市场： A 股

研究频率： 日频横截面选股

自治等级： 受控自治

Alpha 路线： 因子表达式搜索 + 因子组合 + LightGBM/XGBoost

1. 产品定义

Quant Agent 是一个面向 A 股日频横截面选股研究的自治量化研发系统。它以研究目标和约束为输入，自动完成数据获取、数据质检、Alpha 候选生成、模型训练、组合构建、回测、稳健性验证、报告生成以及 Git 提交/PR 创建；Git 合并、生产部署和实盘上线必须由人类审批。

系统定位是“可审计的自动研究工程师”，而不是能够绕过风险控制的自动交易员。LLM 负责提出假设、规划实验、选择工具和解释结果；数据清洗、因子计算、模型训练、回测、风险指标及准入门槛由版本化的确定性程序执行。

1.1 目标

自动发现具有经济逻辑、统计显著性和样本外稳定性的 Alpha。
自动下载并版本化行情、基本面、交易日历和股票状态数据。
自动训练、调参、校准和登记机器学习模型。
自动执行符合 A 股规则且无前视偏差的回测。
自动生成可复现的研究报告、模型卡和部署建议。
自动创建独立分支、提交代码和研究产物、创建 PR。
经人工批准后自动部署到研究或模拟环境。
对每次决策、工具调用、数据和产物提供端到端审计。

1.2 非目标

V1 不支持分钟级、高频、期权、期货或数字资产研究。
V1 不自动合并主分支，不自动生产部署，不自动实盘交易。
不让 LLM 直接计算收益率、风险指标或回测撮合结果。
不以单次最高 Sharpe 作为 Alpha 选择标准。
不允许研究代码直接访问生产交易凭证。

1.3 成功标准

系统成功不等于“保证盈利”，而是满足以下工程与研究标准：

同一代码、配置、数据快照和随机种子可重复得到一致结果。
任何报告指标都能追溯到一次不可变的实验运行。
自动流水线在单节点失败后可从最近检查点恢复，不重复产生副作用。
所有候选策略均经过反泄漏、成本、容量、稳定性和样本外检验。
未经人工审批，系统无法合并 Git、生产部署或连接实盘交易。

2. 用户与使用场景

2.1 角色

角色	权限与职责
Quant Researcher	提交研究目标、查看实验、否决候选、评论报告
Reviewer	审阅代码、数据谱系、报告和模型卡，批准或拒绝 Git 合并
Risk Approver	审核风险、容量、暴露和模拟运行结果，批准生产部署
Platform Admin	管理工具白名单、MCP Server、密钥、配额和审计策略
Quant Agent	在授予的研究权限内执行任务，不拥有审批权

2.2 典型输入

research_goal:
  universe: CSI800
  rebalance_frequency: weekly
  prediction_horizon: 5d
  objective: maximize_rank_ic_after_cost
  benchmark: 000906.SH
  constraints:
    max_industry_active_weight: 0.03
    max_stock_weight: 0.02
    max_turnover_one_way: 0.20
    exclude_st: true
    min_listing_days: 120
    min_adv20_cny: 30000000
  data_cutoff: "2026-06-30T15:00:00+08:00"

2.3 典型输出

可执行的 Alpha 表达式与经济逻辑说明。
数据快照 ID、特征集版本、标签定义和训练/验证/测试切分。
模型文件、模型卡、参数、随机种子和依赖锁文件。
回测结果、交易明细、归因、容量与压力测试。
HTML/PDF/Markdown 研究报告。
Git 分支、提交 SHA、PR 链接和审批包。
经批准后的部署版本、运行状态和回滚目标。

3. 总体架构

flowchart TB
    U[Researcher / Reviewer] --> API[Control Plane API]
    API --> ORCH[Workflow Orchestrator]
    ORCH <--> PLAN[Planner Agent]
    ORCH <--> MEM[Memory Service]
    ORCH <--> POLICY[Policy & Approval Engine]
    ORCH <--> BUS[(Event Bus)]

    BUS --> DATA[Data Agent]
    BUS --> ALPHA[Alpha Discovery Agent]
    BUS --> TRAIN[Training Agent]
    BUS --> BT[Backtest Agent]
    BUS --> REFLECT[Reflection Agent]
    BUS --> REPORT[Report Agent]
    BUS --> GIT[Git Agent]
    BUS --> DEPLOY[Deploy Agent]

    DATA --> MCP[MCP Gateway]
    ALPHA --> MCP
    TRAIN --> MCP
    BT --> MCP
    REPORT --> MCP
    GIT --> MCP
    DEPLOY --> MCP

    MCP --> DS[(Data Lake / Warehouse)]
    MCP --> EXP[(Experiment & Model Registry)]
    MCP --> RUN[Isolated Compute Runners]
    MCP --> GITLAB[Git Provider]
    MCP --> CICD[CI/CD Platform]
    MCP --> OBS[Observability / Audit]

    POLICY -. approval .-> GIT
    POLICY -. approval .-> DEPLOY

3.1 分层原则

控制平面： 接收目标、维护状态机、分配任务、预算和权限。
推理平面： Planner 和各专业 Agent 进行规划、假设生成与解释。
执行平面： 沙箱中的确定性工具运行数据、模型和回测代码。
存储平面： 对象存储、元数据仓库、向量索引、实验与模型注册表。
治理平面： 审批、策略、密钥、审计、成本与资源配额。

3.2 推荐技术栈

能力	推荐实现	可替代实现
API	FastAPI + Pydantic	Go / Spring Boot
工作流	Temporal	Dagster / Prefect
Agent 图	LangGraph（薄层使用）	自研状态机
消息总线	Kafka / Redpanda	Redis Streams
元数据	PostgreSQL	MySQL
对象存储	S3 / MinIO + Parquet	HDFS
特征查询	DuckDB + Polars	Spark
实验/模型注册	MLflow	Weights & Biases
训练	LightGBM、XGBoost、Optuna	CatBoost
回测	自研向量化日频引擎或 Qlib 适配层	Backtrader
报告	Jinja2 + Plotly + WeasyPrint	Quarto
可观测性	OpenTelemetry + Prometheus + Grafana	Datadog
密钥	Vault / 云密钥管理	Kubernetes Secrets
部署	Docker + Kubernetes + Argo CD	Nomad

4. Agent Workflow

4.1 主工作流

flowchart TD
    A[接收研究目标] --> B[策略与权限预检]
    B --> C[Planner 生成 Research Plan]
    C --> D[数据目录检索与缺口分析]
    D --> E[下载 / 更新 / 质检 / 快照]
    E --> F[生成 Alpha 假设与候选表达式]
    F --> G[反泄漏与表达式静态检查]
    G --> H[批量计算因子]
    H --> I[单因子检验与去冗余]
    I --> J[特征集与标签固化]
    J --> K[Walk-forward 训练与调参]
    K --> L[组合构建与约束求解]
    L --> M[事件驱动回测]
    M --> N[稳健性 / 容量 / 压力测试]
    N --> O{Reflection Gate}
    O -->|修订假设| F
    O -->|修订数据/特征| D
    O -->|修订模型| K
    O -->|淘汰| X[归档失败知识]
    O -->|通过| P[生成报告与模型卡]
    P --> Q[创建 Git 分支与提交]
    Q --> R[创建 PR]
    R --> S{人工合并审批}
    S -->|拒绝| O
    S -->|批准| T[CI 构建研究/模拟制品]
    T --> V{生产部署审批}
    V -->|拒绝| Y[保留候选版本]
    V -->|批准| W[部署 + 健康检查 + 可回滚]

4.2 状态机

DRAFT -> PLANNED -> DATA_READY -> DISCOVERING -> TRAINING -> BACKTESTING
      -> REFLECTING -> REPORTING -> AWAITING_MERGE_APPROVAL
      -> MERGED -> AWAITING_DEPLOY_APPROVAL -> DEPLOYING -> DEPLOYED

任意可执行状态 -> RETRY_WAIT -> 原状态
任意可执行状态 -> PAUSED_BY_POLICY
任意状态 -> CANCELLED
失败且不可恢复 -> FAILED

每个状态变更必须写入 workflow_event，包含 run_id、前后状态、原因、触发者、时间、trace ID 和关联产物。

4.3 Agent 职责

Agent	输入	核心职责	输出
Planner	研究目标、预算、历史记忆	分解 DAG、定义验收门槛、分配预算	ResearchPlan
Data	计划、数据目录	选择数据源、增量下载、校验、快照	DatasetSnapshot
Alpha Discovery	数据字典、失败知识	生成经济假设、表达式搜索、初筛	AlphaCandidate[]
Training	特征集、标签、切分	Walk-forward 训练、调参、校准	ModelVersion
Backtest	信号、组合约束、市场规则	撮合、成本、归因、压力测试	BacktestRun
Reflection	全部证据与门槛	诊断失败、决定修订/淘汰/晋级	ReflectionDecision
Report	实验谱系与指标	生成报告、模型卡、审批摘要	ReportBundle
Git	批准的变更包	分支、commit、push、PR；禁止 merge	PullRequest
Deploy	已合并版本、审批令牌	构建、发布、健康检查、回滚	Deployment

4.4 并发和预算

并行只用于彼此独立的候选因子、随机种子、模型配置和压力场景。
每个研究运行必须定义 token、CPU、GPU、内存、存储、API 调用和墙钟时间上限。
触达软限额时 Planner 缩小搜索空间；触达硬限额时状态转为 PAUSED_BY_POLICY。
全局并发、数据源 QPS、训练并发和回测并发由租户及环境配额控制。

5. Memory

5.1 Memory 分层

flowchart LR
    WM[Working Memory\n当前运行上下文] --> EM[Episodic Memory\n历史实验与决策]
    SM[Semantic Memory\n领域知识与数据字典] --> WM
    PM[Procedural Memory\n流程/工具/策略版本] --> WM
    AM[Artifact Memory\n数据/特征/模型/报告] --> EM
    EM --> RET[Hybrid Retrieval]
    SM --> RET
    PM --> RET
    RET --> PLAN[Planner / Agents]

类型	内容	存储	生命周期
Working	当前计划、最近工具结果、预算、待办	Redis/PostgreSQL	单次运行，结束后摘要
Episodic	实验、决策、失败原因、反思	PostgreSQL + 向量索引	长期
Semantic	A 股规则、因子定义、数据字段、研究规范	文档库 + 向量索引	版本化长期
Procedural	Prompt、工具 Schema、工作流、策略规则	Git / Registry	版本化长期
Artifact	Parquet、模型、图表、报告、日志	对象存储	按保留策略

5.2 Memory Record

{
  "memory_id": "mem_01J...",
  "tenant_id": "team_quant",
  "type": "episodic",
  "scope": "alpha_discovery",
  "summary": "20 日反转在小盘股有效，但成本后衰减明显",
  "evidence_refs": ["exp_01J...", "bt_01J..."],
  "tags": ["reversal", "CSI800", "turnover"],
  "valid_from": "2026-07-10T00:00:00Z",
  "knowledge_cutoff": "2026-06-30T07:00:00Z",
  "confidence": 0.78,
  "created_by": "reflection-agent@1.0",
  "policy": {"ttl_days": 730, "contains_secret": false}
}

5.3 检索规则

使用结构化过滤 + BM25 + 向量相似度的混合检索。
必须按租户、项目、市场、频率和 knowledge_cutoff 过滤。
优先检索失败实验和近似候选，防止重复探索。
Memory 只提供建议；关键事实必须回到证据产物验证。
报告、网页或工具输出均视为不可信内容，不得覆盖系统策略或工具权限。

5.4 写入规则

工具原始输出不直接进入长期记忆，只保存引用和经过验证的摘要。
Reflection Agent 只能提出记忆写入建议，由 Memory Policy 校验后落库。
冲突事实并存并标记版本、证据和置信度，不做静默覆盖。
支持按运行、用户或数据源执行删除和保留策略。

6. Planning

6.1 分层计划

Mission Plan： 研究目标、宇宙、预测周期、基准、风险约束和预算。
Research Plan： 数据、假设族、标签、切分、检验、模型和晋级门槛。
Execution DAG： 可重试的工具任务、依赖、输入输出和幂等键。
Next Action： Agent 每轮只能选择一个有明确 Schema 的动作。

6.2 ResearchPlan Schema

{
  "plan_id": "plan_01J...",
  "objective": "发现 CSI800 中预测未来 5 日超额收益的稳健信号",
  "as_of": "2026-06-30T15:00:00+08:00",
  "universe": "CSI800_POINT_IN_TIME",
  "hypothesis_families": ["reversal", "quality", "liquidity", "interaction"],
  "label": "next_5d_vwap_to_vwap_excess_return",
  "validation": {
    "scheme": "purged_walk_forward",
    "train_years": 5,
    "validation_months": 12,
    "test_months": 12,
    "embargo_days": 5
  },
  "gates": {
    "min_oos_rank_ic": 0.02,
    "min_icir": 0.5,
    "min_net_sharpe": 1.0,
    "max_drawdown": 0.20,
    "max_one_way_turnover": 0.20
  },
  "budget": {
    "max_candidates": 500,
    "max_training_jobs": 80,
    "max_wall_clock_hours": 8,
    "max_cost_cny": 500
  }
}

6.3 Planner 约束

计划必须在任何数据下载或训练前冻结 as_of 和数据可得时间规则。
每个任务声明前置条件、预期产物、验证器、超时、重试和补偿动作。
修改标签、宇宙、数据切分或主要回测假设会生成新的 plan_revision。
Planner 无权降低组织级硬门槛；只能提高门槛或请求人工例外。
计划变化必须说明新证据、影响范围和已消耗预算。

7. Alpha 自动发现

7.1 搜索空间

原子字段： OHLCV、成交额、换手率、市值、估值、财务质量、分析师一致预期（若有可靠 PIT 数据）。
时间算子： delay、delta、rolling mean/std/min/max/rank/corr、decay。
横截面算子： rank、zscore、winsorize、neutralize、group rank。
组合算子： 加减乘除、条件、截断；限制表达式深度和复杂度。
模型： 线性基线、LightGBM、XGBoost；深度模型不进入 V1。

7.2 候选生成

从经济假设模板、历史失败知识和数据字典生成候选。
符号搜索使用语法树而非任意 Python 代码。
候选先经过类型、单位、可得时间、复杂度和重复性静态检查。
以多目标优化选择候选：样本外 IC、稳定性、换手、复杂度、与已有因子相关性。

7.3 防止数据窥探

研究运行开始前登记候选生成空间和最大试验次数。
记录全部候选，包括失败候选，估计多重检验压力。
使用 Deflated Sharpe Ratio、Probability of Backtest Overfitting 或等价方法。
最终锁箱测试集只能在候选冻结后运行；同一研究任务默认只允许解锁一次。
任何读取测试期结果后发生的修改都创建新研究世代，不能沿用原测试结论。

8. 数据系统

8.1 数据源

V1 推荐优先使用 Tushare Pro 或合规商业数据源，并为交易所公开数据提供校验适配器。必须确认数据许可允许自动下载、缓存和内部衍生使用。

必需数据集：

交易日历、股票主数据、上市/退市日期、名称与状态变更。
日线未复权行情、复权因子、停牌、涨跌停价格。
指数成分及权重的历史时点版本，禁止使用当前成分回填历史。
财务数据及公告日期，采用 point-in-time 可得性。
ST/*ST、风险警示、行业分类的历史变化。

8.2 数据分层

RAW       原始响应，不修改，可校验 checksum
CURATED   类型统一、去重、主键和质量规则通过
PIT       按实际可得时间对齐的时点数据
FEATURE   版本化特征和标签
SERVING   研究/模拟环境读取的发布快照

8.3 快照与谱系

每个 dataset_snapshot_id 固定以下内容：数据源版本、请求参数、下载时间、覆盖区间、对象 checksum、Schema、质量报告、可得时间规则和上游快照。任何训练、回测或报告只引用快照 ID，不引用可变的“最新表”。

8.4 数据质量门槛

主键唯一性、Schema、空值率、覆盖率、数值范围和跨源一致性。
交易日缺口、复权连续性、价格/成交量异常跳变。
公告日在报告期之前、未来成分泄漏、退市样本缺失等专门检查。
质量硬失败阻断流水线；软异常进入报告并降低置信度。

9. 模型训练

9.1 标签与切分

默认标签：未来 5 个交易日 VWAP-to-VWAP 相对基准或行业的收益。
特征必须滞后到决策时刻真实可得；财务字段以公告日对齐。
使用 Purged Walk-forward，purge/embargo 至少覆盖标签持有期。
预处理器只在训练窗口拟合，并与模型一同版本化。

9.2 训练流程

训练线性或简单排序基线。
LightGBM/XGBoost 使用受限搜索空间进行 Optuna 调参。
优化目标综合 Rank IC、ICIR、稳定性和预估换手，而非训练损失单指标。
多随机种子训练并报告均值、方差和最差结果。
用 SHAP、置换重要性和单调性检查解释模型。
登记模型、依赖、特征、数据、代码 SHA、参数和指标。

9.3 模型准入

必须优于简单基线，并在多数 walk-forward 窗口保持方向一致。
不允许任何单特征支配模型且缺乏经济解释。
训练/验证性能差距超过策略阈值时标记过拟合并阻断晋级。
模型文件必须来自隔离训练任务并通过恶意反序列化安全策略；优先使用安全格式。

10. 回测规格

10.1 时间语义

默认信号在 T 日收盘数据完整后生成，订单最早在 T+1 执行。报告必须明确决策时刻、下单时刻、成交价格假设和持有周期。

10.2 A 股市场规则

股票交易遵循 T+1 卖出约束。
停牌期间不能成交。
涨停时买单不得假设成交，跌停时卖单不得假设成交；可配置更保守的成交概率模型。
使用历史 ST 状态、上市天数、退市和指数成分。
佣金、印花税、过户费按交易方向和历史生效日期配置。
滑点使用成交额参与率、ADV 和冲击函数建模。
禁止用后复权价格直接模拟现金交易；收益计算和公司行动需保持一致。

10.3 组合构建

模型分数经行业/市值中性化与异常值处理后进入优化器。
约束包括单股权重、行业主动权重、风格暴露、换手、流动性和持仓数。
优化失败时使用确定性降级策略，并在报告中标记。
目标组合到实际成交之间保留未成交订单和现金状态。

10.4 指标

收益：年化收益、超额收益、胜率、月度/年度分布。
风险：波动率、Sharpe、Sortino、最大回撤、Calmar、VaR/ES。
信号：Pearson/Spearman IC、ICIR、分层收益、单调性、衰减。
交易：换手、成本、滑点、成交率、容量、平均持有期。
暴露：行业、市值、Beta、风格和集中度。
稳健性：分时期、牛熊市、行业、规模、参数扰动、成本倍增。

10.5 建议晋级门槛

这些是组织可配置的初始值，不是收益承诺：

维度	建议门槛
样本外 Rank IC	>= 0.02
样本外 ICIR	>= 0.5
成本后 Sharpe	>= 1.0
最大回撤	<= 20%
年化单边换手	与策略约束一致且成本后仍通过
窗口稳定性	至少 70% OOS 窗口 IC 为正
成本压力	2 倍基准成本下不出现结论性崩溃
容量	目标资金规模下参与率不超限
泄漏测试	100% 通过

11. Reflection

11.1 Reflection Loop

flowchart LR
    E[Evidence Bundle] --> C[Critic: 检查证据完整性]
    C --> D[Diagnoser: 定位失败类型]
    D --> J{Judge}
    J -->|Revise Data| RD[数据修订]
    J -->|Revise Alpha| RA[假设修订]
    J -->|Revise Model| RM[训练修订]
    J -->|Reject| R[归档失败知识]
    J -->|Promote| P[生成审批包]
    RD --> E
    RA --> E
    RM --> E

11.2 反思输入

计划与所有 revision。
数据质量和可得时间报告。
候选全集、试验次数和多重检验统计。
训练、验证、测试、回测、归因和压力测试结果。
工具错误、重试、资源消耗和策略违规。
与历史相似实验及失败记忆的对比。

11.3 决策分类

决策	条件	后续动作
RETRY	临时性工具/资源故障	使用同一幂等键重试
REVISE_DATA	缺失、漂移、PIT 或质量问题	回到数据阶段，增加 plan revision
REVISE_ALPHA	逻辑弱、冗余、衰减、成本过高	修改候选族，不得读取锁箱后继续微调
REVISE_MODEL	过拟合、不稳定、解释异常	缩小复杂度或改变训练方案
REJECT	硬门槛失败或预算耗尽	归档证据和可复用失败结论
PROMOTE	全部门槛通过且证据完整	生成报告和审批包

11.4 防止无限循环

每个阶段设最大 revision 次数和总预算。
同类失败连续两次后必须改变假设或终止，不能只换随机种子。
Reflection 不能自行修改硬门槛或扩大预算。
所有晋级结论必须引用机器可验证的 Gate Result。

12. Tool Calling

12.1 调用链

sequenceDiagram
    participant A as Agent
    participant G as Tool Gateway
    participant P as Policy Engine
    participant M as MCP Server
    participant W as Worker
    participant O as Audit Store

    A->>G: tool_call(name, args, context)
    G->>G: Schema 校验 + 幂等键
    G->>P: authorize(subject, action, resource)
    P-->>G: allow / deny / approval_required
    G->>M: MCP tools/call
    M->>W: 隔离任务
    W-->>M: result + artifact refs
    M-->>G: structured result
    G->>O: append audit event
    G-->>A: verified result / typed error

12.2 工具规范

每个工具必须声明：

名称、版本、描述和 JSON Schema。
read、compute、write、external_side_effect 风险级别。
所需 scope、资源上限、超时和最大返回大小。
是否幂等、幂等键组成、重试策略和补偿动作。
输入/输出中的敏感字段和日志脱敏策略。
可验证器，例如 checksum、行数、指标范围或签名。

12.3 通用调用信封

{
  "request_id": "req_01J...",
  "run_id": "run_01J...",
  "plan_id": "plan_01J...",
  "agent": "data-agent@1.0",
  "tool": "market_data.fetch_daily@1.0",
  "arguments": {},
  "idempotency_key": "sha256(...) ",
  "deadline": "2026-07-10T10:30:00Z",
  "scopes": ["data:read", "artifact:write"],
  "trace_id": "00-..."
}

12.4 错误模型

{
  "ok": false,
  "error": {
    "code": "DATA_RATE_LIMITED",
    "category": "transient",
    "message": "Provider quota exceeded",
    "retry_after_seconds": 60,
    "details_ref": "artifact://logs/..."
  }
}

错误分类为 validation、policy、transient、permanent、data_quality 和 internal。仅 transient 可自动指数退避重试；具有外部副作用的工具在重试前必须查询幂等状态。

12.5 沙箱

代码执行使用无生产凭证的临时容器，默认禁网、只读基础镜像和非 root 用户。
数据访问通过受控工具或短期签名 URL，不挂载完整密钥。
限制 CPU、内存、磁盘、进程数、执行时间和输出大小。
依赖只能来自锁定仓库和 allowlist；生成 SBOM 并扫描漏洞。

13. MCP 接口

13.1 MCP 拓扑

Agent 只连接内部 MCP Gateway。Gateway 根据工具命名空间路由到独立 Server，并统一执行认证、授权、限流、Schema 校验、审计和结果截断。

mcp-gateway
├── market-data-mcp
├── feature-store-mcp
├── experiment-mcp
├── backtest-mcp
├── artifact-mcp
├── git-mcp
├── deploy-mcp
└── observability-mcp

13.2 Resources

URI 模板	含义
`dataset://catalog/{dataset}`	数据集 Schema、覆盖、许可和新鲜度
`dataset://snapshot/{snapshot_id}`	不可变数据快照元数据
`feature://definition/{feature_id}`	因子表达式、依赖和版本
`experiment://run/{run_id}`	实验参数、指标和谱系
`model://version/{model_id}`	模型卡、签名和阶段
`backtest://run/{backtest_id}`	回测配置、指标和交易产物
`artifact://sha256/{digest}`	内容寻址产物元数据
`policy://gate/{gate_name}`	当前门槛和策略版本
`git://repo/{repo}/pr/{number}`	PR 状态和检查结果
`deploy://environment/{env}`	环境版本与健康状态

13.3 Prompts

Prompt	用途
`plan_quant_research`	将研究目标转换为 ResearchPlan 草案
`generate_alpha_hypotheses`	基于字段、约束和失败知识生成候选假设
`critique_experiment`	对证据包执行结构化批评，不计算指标
`summarize_approval_package`	为 Reviewer 生成带证据引用的摘要

Prompt 只是版本化模板，不授予额外工具权限。

13.4 核心 Tools

market-data-mcp

market_data.catalog(query) -> DatasetDescriptor[]
market_data.fetch_daily(symbols, start, end, fields, as_of) -> JobRef
market_data.fetch_fundamentals(table, periods, as_of) -> JobRef
market_data.validate(snapshot_id, ruleset) -> DataQualityReport
market_data.create_snapshot(job_refs, schema_version) -> DatasetSnapshot

fetch_daily 输入 Schema 示例：

{
  "type": "object",
  "required": ["symbols", "start", "end", "fields", "as_of"],
  "properties": {
    "symbols": {"type": "array", "items": {"type": "string"}, "maxItems": 5000},
    "start": {"type": "string", "format": "date"},
    "end": {"type": "string", "format": "date"},
    "fields": {"type": "array", "items": {"enum": ["open", "high", "low", "close", "volume", "amount", "adj_factor"]}},
    "as_of": {"type": "string", "format": "date-time"},
    "adjustment": {"enum": ["none", "forward", "backward"], "default": "none"}
  },
  "additionalProperties": false
}

feature-store-mcp

feature.validate_expression(ast, availability_policy) -> ValidationReport
feature.compute(snapshot_id, expressions, date_range) -> FeatureJob
feature.profile(feature_set_id) -> FeatureProfile
feature.register(definitions, evidence_refs) -> FeatureSetVersion

experiment-mcp

experiment.create(plan_id, lineage) -> ExperimentRun
experiment.train(feature_set_id, label_id, split, model_spec) -> TrainingJob
experiment.get_metrics(job_id) -> MetricBundle
model.register(training_job_id, stage="candidate") -> ModelVersion
model.compare(model_ids, gates) -> ComparisonResult

backtest-mcp

backtest.validate_config(config) -> ValidationReport
backtest.run(model_id, snapshot_id, portfolio_spec, execution_spec) -> BacktestJob
backtest.stress(backtest_id, scenarios) -> StressTestBundle
backtest.get_gate_results(backtest_id, policy_version) -> GateResult[]

backtest.run 必须要求显式传入：信号可得时间、成交价格、费用版本、涨跌停规则、停牌规则、T+1、基准、组合约束和随机种子（如适用）。

artifact-mcp

artifact.put(content_type, checksum, payload_or_upload_ref) -> ArtifactRef
artifact.get_metadata(uri) -> ArtifactMetadata
artifact.sign_manifest(artifact_refs, lineage) -> SignedManifest
report.render(template_version, evidence_bundle) -> ReportBundle

git-mcp

git.create_branch(repo, base_sha, branch_name) -> BranchRef
git.apply_patch(repo, branch, patch, expected_tree_sha) -> ChangeSet
git.run_checks(repo, branch, check_suite) -> CheckRun
git.commit(repo, branch, message, signed_manifest) -> CommitRef
git.push(repo, branch) -> PushResult
git.create_pull_request(repo, branch, base, title, body) -> PullRequest
git.get_pull_request(repo, number) -> PullRequest

V1 不向 Agent 暴露 merge_pull_request、force_push、删除分支、修改保护规则等工具。

deploy-mcp

deploy.build(commit_sha, target, sbom=true) -> BuildArtifact
deploy.plan(artifact_id, environment) -> DeploymentPlan
deploy.request_approval(plan_id) -> ApprovalRequest
deploy.execute(plan_id, approval_token) -> Deployment
deploy.health(deployment_id) -> HealthReport
deploy.rollback(deployment_id, approval_token) -> Deployment

deploy.execute 必须验证短期、单次、绑定 plan_id + artifact_digest + environment 的审批令牌。Agent 不能签发审批令牌。

observability-mcp

audit.append(event) -> EventRef
trace.get(trace_id) -> Trace
workflow.get_status(run_id) -> WorkflowStatus
workflow.cancel(run_id, reason) -> WorkflowStatus
cost.get(run_id) -> CostReport

13.5 MCP 安全要求

使用 OAuth 2.1 或工作负载身份，短期 token 和细粒度 scope。
Server 端执行授权，不能信任 Agent 声称的权限。
对外部内容做 prompt injection 隔离；资源内容不能成为系统指令。
工具输入输出全量审计，但密钥、token、个人信息必须脱敏。
MCP Server 版本、工具 Schema 和策略版本进入每次运行清单。

14. Git 自动化

14.1 分支与提交

分支名：agent/{run_id}/{slug}。
仅允许修改策略白名单目录，例如 research/、configs/、reports/ 和生成的测试。
禁止修改 CI 权限、分支保护、密钥、审批规则和生产基础设施所有权文件。
commit 必须签名，并包含 run_id、实验 ID、数据快照 ID 和报告链接。
所有自动生成文件经过格式化、静态检查、单元测试、泄漏测试和依赖扫描。

14.2 PR 模板

## Research objective
## Economic hypothesis
## Data and point-in-time assumptions
## Experiment lineage
## Out-of-sample results
## Costs, capacity and stress tests
## Known limitations
## Reproduction command
## Approval checklist

14.3 审批门

Reviewer 必须是 Agent 之外的人类身份。
CODEOWNERS 至少要求 Quant Reviewer；涉及部署配置时额外要求 Risk/Platform Reviewer。
PR 检查未通过、提交 SHA 变化或报告清单不匹配会使已有批准失效。

15. 部署

15.1 环境

research -> simulation -> paper -> production

V1 可自动部署到隔离 research/simulation 环境；paper/production 必须人工批准。实盘交易接口不在 V1 工具集合中。

15.2 发布流程

基于已合并 commit 构建不可变镜像。
生成 SBOM、漏洞扫描、镜像签名和 provenance。
验证模型、特征、配置与报告清单的一致性。
在 simulation 执行 smoke test 和历史重放。
创建部署计划和差异摘要。
人工审批生产计划。
灰度发布、健康检查和数据新鲜度检查。
异常时自动停止并回滚到已批准版本。

15.3 运行时监控

数据新鲜度、Schema 漂移、缺失率和异常值。
特征分布 PSI、模型分数漂移、预测覆盖率。
模拟收益、换手、成本、暴露和约束违规。
服务延迟、错误率、资源和费用。
漂移触发重新研究请求，但不能自动越过审批上线新模型。

16. 安全、权限与治理

16.1 权限矩阵

动作	Agent 自动	人工批准后	禁止
读取已授权研究数据	是
下载并缓存合规数据	是
运行沙箱代码/训练/回测	是
写研究产物和候选模型	是
创建分支、commit、push、PR	是
合并受保护分支		是
部署 research/simulation	是（策略可配置）
部署 paper/production		是
发出实盘订单			V1 禁止
修改审批/分支保护/密钥策略			Agent 禁止

16.2 威胁控制

Prompt injection：外部数据标记为 untrusted，工具输出不能修改系统策略。
数据投毒：来源白名单、checksum、跨源校验、异常检测和数据快照签名。
供应链：依赖锁定、SBOM、签名镜像、禁止运行未知 pickle。
凭证泄漏：Vault 动态凭证、最小权限、日志脱敏、禁止凭证进入上下文。
越权：策略引擎在工具网关和 MCP Server 双重执行。
审计篡改：append-only 审计存储、哈希链、定期归档。

17. 可观测性与审计

每次运行必须关联统一的 trace_id，并记录：

输入目标、计划版本、Agent/Prompt/模型版本。
每次工具调用的参数摘要、授权结果、耗时、费用和结果引用。
数据、特征、模型、回测、报告、代码和部署的完整谱系。
Reflection 决策、Gate Result、人工批准者和批准对象摘要。
token、API、计算、存储和数据源费用。

关键 SLO：

控制平面月可用性 >= 99.5%。
工作流状态事件不丢失，恢复点目标 RPO 近零。
95% 可重试失败在策略时限内自动恢复。
100% 生产部署可追溯到人工批准和不可变制品摘要。

18. 核心数据模型

erDiagram
    RESEARCH_RUN ||--o{ PLAN_REVISION : has
    RESEARCH_RUN ||--o{ TOOL_CALL : executes
    RESEARCH_RUN ||--o{ EXPERIMENT_RUN : contains
    DATASET_SNAPSHOT ||--o{ FEATURE_SET : feeds
    FEATURE_SET ||--o{ EXPERIMENT_RUN : trains
    EXPERIMENT_RUN ||--o{ MODEL_VERSION : produces
    MODEL_VERSION ||--o{ BACKTEST_RUN : evaluated_by
    BACKTEST_RUN ||--o{ GATE_RESULT : yields
    RESEARCH_RUN ||--o{ REFLECTION : records
    RESEARCH_RUN ||--o{ REPORT_BUNDLE : produces
    REPORT_BUNDLE ||--o| PULL_REQUEST : supports
    PULL_REQUEST ||--o| DEPLOYMENT : releases

所有实体采用不可变 ID；修订通过新版本和父引用表达，不在原记录上静默覆盖。

19. API 草案

POST   /v1/research-runs                 创建研究运行
GET    /v1/research-runs/{run_id}        查询状态与当前计划
POST   /v1/research-runs/{run_id}/pause  暂停
POST   /v1/research-runs/{run_id}/resume 恢复
POST   /v1/research-runs/{run_id}/cancel 取消
GET    /v1/research-runs/{run_id}/lineage 获取谱系
GET    /v1/research-runs/{run_id}/report 获取报告
POST   /v1/approvals                     提交人类审批决定
GET    /v1/approvals/{approval_id}       查询审批
GET    /v1/artifacts/{artifact_id}       查询产物元数据

所有写请求支持幂等键；API 返回操作引用，长任务异步执行。

20. 测试与验收

20.1 测试层级

单元测试：因子算子、时间对齐、费用、撮合、指标、策略规则。
属性测试：未来数据变化不得影响过去信号；现金和持仓守恒。
Golden Test：固定小数据集上的信号、交易和净值结果固定。
集成测试：各 MCP Server 的 Schema、权限、幂等和错误模型。
端到端测试：从研究目标到 PR，以及审批后的 simulation 部署。
安全测试：越权、prompt injection、恶意数据、依赖和密钥泄漏。
故障测试：数据源限流、Worker 中断、消息重复、部署健康失败。

20.2 关键验收场景

修改 T+5 的数据不会改变 T 日信号或 T+1 订单。
当股票 T+1 涨停时，买单不产生不现实成交。
当数据源返回重复或缺失交易日时，流水线被质量门阻断。
同一幂等键重复下载、提交或部署不会产生重复副作用。
Agent 调用 Git merge 或 production deploy（无审批令牌）必须被拒绝并审计。
报告中的任一数字能定位到回测产物、配置、代码和数据快照。
Worker 中断后恢复不会改变实验 ID 或重复登记模型。
Prompt injection 文本无法改变工具白名单或审批策略。
PR 的 commit SHA 变化后原审批失效。
部署健康检查失败时回滚到明确的上一批准版本。

20.3 Definition of Done

所有验收场景自动化并通过。
完成端到端数据与实验谱系演示。
完成威胁模型、权限审计和恢复演练。
在固定基准任务上成功生成可复现报告和 PR。
未经人工审批的生产部署通过技术手段不可达。

21. 交付阶段

Phase 0：基础治理

定义数据许可、研究模板、回测假设、门槛、权限、审批和审计格式。

Phase 1：可复现研究底座

完成数据快照、特征表达式、训练、A 股回测、实验注册和报告，不引入 Agent 自主规划。

Phase 2：单 Agent 编排

引入 Planner、工具网关、Memory 和 Reflection，自动完成一次受限研究任务。

Phase 3：专业 Agent 与 MCP

拆分 Data/Alpha/Training/Backtest/Report Agent，部署 MCP Gateway 和专业 Server。

Phase 4：Git 与部署自动化

自动创建 commit/PR，接入人工审批，支持 simulation 自动部署和 production 审批发布。

Phase 5：规模化与治理

增加并行搜索、成本控制、漂移监控、多团队隔离和审计报表。

22. 待组织确认的配置项

以下内容不阻塞架构，但在实施前必须由业务和合规共同确认：

合规数据供应商、许可范围、历史深度和预算。
目标股票池、基准、资金规模和换手/容量约束。
费用和滑点模型的组织标准。
研究晋级门槛及例外审批流程。
Git Provider、CI/CD、云环境和身份系统。
报告保留期限、审计保留期限和模型风险分级。

23. 设计决策摘要

采用分层 Agent + 确定性工作流，而不是自由多 Agent 对话。
采用受控自治：研究自动化，合并、生产部署和实盘必须人审。
V1 聚焦 A 股日频横截面，避免同时支持多市场和高频。
Alpha 使用可解释因子搜索与树模型组合，不引入深度模型。
所有结论由版本化证据支撑，LLM 不作为指标计算器或审批者。
MCP 是受策略保护的工具边界，不是绕过权限的通道。
数据时点、失败实验和多重检验被视为一等公民。

NautilusTrader第7章：技术指标系统

Henry Lin — Fri, 10 Jul 2026 01:06:51 +0000

第7章：技术指标系统

学习目标

通过本章学习，您将：

掌握 NautilusTrader 内置技术指标的使用
理解指标的更新机制和生命周期
学会创建自定义技术指标
实现指标级联和组合
优化指标计算性能

7.1 技术指标概览

7.1.1 指标基类

NautilusTrader 的所有指标都继承自 Indicator 基类：

"""
技术指标基类说明
"""

from abc import ABC, abstractmethod
from decimal import Decimal
from typing import Any, Optional

from nautilus_trader.model.data import Bar
from nautilus_trader.core.message import Event


class Indicator(ABC):
    """技术指标基类"""

    def __init__(
        self,
        params: Optional[dict] = None,
        name: Optional[str] = None,
    ):
        """
        初始化指标

        Parameters
        ----------
        params : dict, optional
            指标参数
        name : str, optional
            指标名称
        """
        self._params = params or {}
        self._name = name or self.__class__.__name__
        self._initialized = False
        self._has_inputs = False
        self._inputs = 0

    @property
    def name(self) -> str:
        """获取指标名称"""
        return self._name

    @property
    def initialized(self) -> bool:
        """是否已初始化"""
        return self._initialized

    @property
    def has_inputs(self) -> bool:
        """是否有输入数据"""
        return self._has_inputs

    @property
    def count_inputs(self) -> int:
        """输入数据数量"""
        return self._inputs

    @abstractmethod
    def handle_bar(self, bar: Bar) -> None:
        """
        处理K线数据（抽象方法）

        Parameters
        ----------
        bar : Bar
            K线数据
        """
        pass

    @abstractmethod
    def handle_quote_tick(self, quote) -> None:
        """
        处理报价数据（抽象方法）

        Parameters
        ----------
        quote : QuoteTick
            报价数据
        """
        pass

    @abstractmethod
    def handle_trade_tick(self, trade) -> None:
        """
        处理成交数据（抽象方法）

        Parameters
        ----------
        trade : TradeTick
            成交数据
        """
        pass

    @abstractmethod
    def reset(self) -> None:
        """重置指标状态"""
        pass

    @abstractmethod
    def _update(self, value: float) -> None:
        """更新指标值（内部方法）"""
        pass

    def update_raw(self, value: float) -> None:
        """
        更新原始值

        Parameters
        ----------
        value : float
            原始值
        """
        if not isinstance(value, (int, float, Decimal)):
            raise TypeError(f"Invalid input type: {type(value)}")

        self._has_inputs = True
        self._inputs += 1
        self._update(float(value))

    def to_dict(self) -> dict:
        """转换为字典"""
        return {
            'name': self.name,
            'params': self._params,
            'initialized': self.initialized,
            'inputs': self._inputs,
            'has_inputs': self.has_inputs,
        }


# 自定义指标示例
class CustomIndicator(Indicator):
    """自定义指标示例"""

    def __init__(self, period: int = 20):
        """
        初始化指标

        Parameters
        ----------
        period : int
            计算周期
        """
        super().__init__(
            params={'period': period},
            name="CustomIndicator"
        )

        self.period = period
        self.values = []  # 存储历史值
        self._value = 0.0

    @property
    def value(self) -> float:
        """获取指标值"""
        return self._value

    @property
    def value_1(self) -> float:
        """获取前一个值"""
        return self.values[-2] if len(self.values) >= 2 else 0.0

    def handle_bar(self, bar: Bar) -> None:
        """处理K线数据"""
        self.update_raw(float(bar.close))

    def handle_quote_tick(self, quote) -> None:
        """处理报价数据"""
        # 使用中间价
        mid_price = (float(quote.bid_price) + float(quote.ask_price)) / 2
        self.update_raw(mid_price)

    def handle_trade_tick(self, trade) -> None:
        """处理成交数据"""
        self.update_raw(float(trade.price))

    def reset(self) -> None:
        """重置指标"""
        self.values.clear()
        self._value = 0.0
        self._initialized = False
        self._has_inputs = False
        self._inputs = 0

    def _update(self, value: float) -> None:
        """更新指标值"""
        self.values.append(value)

        if len(self.values) > self.period:
            self.values.pop(0)

        # 计算简单移动平均
        if len(self.values) >= self.period:
            self._value = sum(self.values) / self.period
            self._initialized = True

7.1.2 内置指标库

NautilusTrader 提供了丰富的内置指标：

"""
内置指标库导览
"""

# 移动平均线指标
from nautilus_trader.indicators.average import (
    SMA,      # 简单移动平均
    EMA,      # 指数移动平均
    WMA,      # 加权移动平均
    DEMA,     # 双指数移动平均
    TEMA,     # 三指数移动平均
    KAMA,     # 考夫曼自适应移动平均
    MAMA,     # MESA自适应移动平均
    FRAMA,    # 分形自适应移动平均
    HMA,      # 赫尔移动平均
)

# 动量指标
from nautilus_trader.indicators.momentum import (
    RSI,      # 相对强弱指数
    MACD,     # MACD指标
    Stochastic, # 随机指标
    CCI,      # 商品通道指数
    ROC,      # 变化率
    MOMENTUM, # 动量指标
    WILLIAMS_R, # 威廉指标
)

# 趋势指标
from nautilus_trader.indicators.trend import (
    ADX,      # 平均趋向指数
    Aroon,    # 阿隆指标
    ATR,      # 平均真实波幅
    ParabolicSAR, # 抛物线SAR
    TRIX,     # 三重指数平滑平均
    Vortex,   # 涡流指标
)

# 波动率指标
from nautilus_trader.indicators.volatility import (
    BollingerBands, # 布林带
    KeltnerChannels, # 肯特纳通道
    DonchianChannels, # 唐奇安通道
    StandardDeviation, # 标准差
)

# 成交量指标
from nautilus_trader.indicators.volume import (
    OBV,      # 能量潮
    VWAP,     # 成交量加权平均价
    MoneyFlowIndex, # 资金流量指数
    AccumulationDistribution, # 累积派发线
    EaseOfMovement, # 简易波动指标
)

# 模糊逻辑指标
from nautilus_trader.indicators.fuzzy_candlesticks import (
    FuzzyCandlestickPattern, # 模糊K线形态
)

7.2 移动平均线指标

7.2.1 基础移动平均

"""
移动平均线使用示例
"""

from decimal import Decimal
from nautilus_trader.model.data import Bar
from nautilus_trader.model.data import BarType
from nautilus_trader.model.enums import BarAggregation, PriceType
from nautilus_trader.model.identifiers import InstrumentId
from nautilus_trader.trading.strategy import Strategy
from nautilus_trader.common.enums import LogColor


class MovingAverageStrategy(Strategy):
    """移动平均线策略"""

    def __init__(
        self,
        bar_type: BarType,
        fast_period: int = 10,
        slow_period: int = 30,
    ):
        """
        初始化策略

        Parameters
        ----------
        bar_type : BarType
            K线类型
        fast_period : int
            快速均线周期
        slow_period : int
            慢速均线周期
        """
        super().__init__()

        self.bar_type = bar_type

        # 创建移动平均线指标
        self.sma_fast = SMA(period=fast_period)
        self.sma_slow = SMA(period=slow_period)

        # 也可以使用EMA
        # self.ema_fast = EMA(period=fast_period)
        # self.ema_slow = EMA(period=slow_period)

        # 状态跟踪
        self.last_signal = None
        self.position = None

    def on_start(self):
        """策略启动"""
        self.log.info("移动平均线策略启动")
        self.log.info(f"快线周期: {self.sma_fast.period}, "
                     f"慢线周期: {self.sma_slow.period}")

        # 订阅K线数据
        self.subscribe_bars(self.bar_type)

    def on_bar(self, bar: Bar):
        """处理K线数据"""
        # 更新指标
        self.sma_fast.update_raw(float(bar.close))
        self.sma_slow.update_raw(float(bar.close))

        # 等待指标初始化
        if not (self.sma_fast.initialized and self.sma_slow.initialized):
            return

        # 获取指标值
        fast_value = self.sma_fast.value
        slow_value = self.sma_slow.value

        # 检查金叉/死叉
        self._check_crossover(fast_value, slow_value, bar)

        # 记录指标值
        if self.count_bars() % 10 == 0:  # 每10根K线记录一次
            self.log.info(
                f"MA - 快线: {fast_value:.2f}, "
                f"慢线: {slow_value:.2f}, "
                f"收盘: {bar.close:.2f}",
                color=LogColor.BLUE
            )

    def _check_crossover(self, fast: float, slow: float, bar: Bar):
        """检查均线交叉"""
        fast_prev = self.sma_fast.value_1
        slow_prev = self.sma_slow.value_1

        # 金叉：快线上穿慢线
        if (fast > slow and fast_prev <= slow_prev and
            self.position != "LONG"):
            self._enter_long(bar)
            self.last_signal = "GOLDEN_CROSS"

        # 死叉：快线下穿慢线
        elif (fast < slow and fast_prev >= slow_prev and
              self.position == "LONG"):
            self._exit_long(bar)
            self.last_signal = "DEATH_CROSS"

    def _enter_long(self, bar: Bar):
        """开多头仓位"""
        self.position = "LONG"
        self.log.info(
            f"金叉信号 - 买入价: {bar.close:.2f}",
            color=LogColor.GREEN
        )
        # 这里应该执行实际的买入操作
        # buy_order = MarketOrder(...)
        # self.submit_order(buy_order)

    def _exit_long(self, bar: Bar):
        """平多头仓位"""
        self.position = None
        self.log.info(
            f"死叉信号 - 卖出价: {bar.close:.2f}",
            color=LogColor.RED
        )
        # 这里应该执行实际的卖出操作
        # sell_order = MarketOrder(...)
        # self.submit_order(sell_order)


# 高级移动平均线策略示例
class AdvancedMAStrategy(Strategy):
    """高级移动平均线策略"""

    def __init__(self, bar_type: BarType):
        """初始化策略"""
        super().__init__()
        self.bar_type = bar_type

        # 多种移动平均线
        self.sma_short = SMA(period=5)
        self.sma_medium = SMA(period=20)
        self.sma_long = SMA(period=50)

        # EMA用于趋势确认
        self.ema_trend = EMA(period=100)

        # 加权移动平均
        self.wma_smooth = WMA(period=10)

        # 自适应移动平均
        self.kama_fast = KAMA(
            fast_period=2,
            slow_period=30,
            period=10
        )

        # 状态变量
        self.trend_strength = 0
        self.market_state = "NEUTRAL"

    def on_start(self):
        """策略启动"""
        self.log.info("高级MA策略启动")
        self.subscribe_bars(self.bar_type)

    def on_bar(self, bar: Bar):
        """处理K线数据"""
        price = float(bar.close)

        # 更新所有指标
        self.sma_short.update_raw(price)
        self.sma_medium.update_raw(price)
        self.sma_long.update_raw(price)
        self.ema_trend.update_raw(price)
        self.wma_smooth.update_raw(price)
        self.kama_fast.update_raw(price)

        # 等待指标初始化
        if not (self.sma_long.initialized and self.ema_trend.initialized):
            return

        # 分析市场状态
        self._analyze_market_state()

        # 生成交易信号
        signal = self._generate_signal()

        if signal:
            self.log.info(
                f"交易信号: {signal['type']} "
                f"(强度: {signal['strength']:.2f})",
                color=LogColor.CYAN
            )

    def _analyze_market_state(self):
        """分析市场状态"""
        short = self.sma_short.value
        medium = self.sma_medium.value
        long = self.sma_long.value

        # 计算均线排列
        if short > medium > long:
            self.market_state = "BULLISH"
        elif short < medium < long:
            self.market_state = "BEARISH"
        else:
            self.market_state = "NEUTRAL"

        # 计算趋势强度
        spread_long_short = abs(short - long) / long
        self.trend_strength = min(spread_long_short * 10, 1.0)

    def _generate_signal(self) -> Optional[dict]:
        """生成交易信号"""
        if self.market_state == "NEUTRAL":
            return None

        # 使用KAMA作为确认
        kama_value = self.kama_fast.value
        current_price = self.sma_short.value

        # 多重确认
        conditions = []

        # 1. 市场状态
        if self.market_state == "BULLISH":
            conditions.append(kama_value > current_price * 0.995)  # 接近或高于
        elif self.market_state == "BEARISH":
            conditions.append(kama_value < current_price * 1.005)  # 接近或低于

        # 2. 趋势强度
        conditions.append(self.trend_strength > 0.3)

        # 3. EMA趋势确认
        if self.market_state == "BULLISH":
            conditions.append(self.ema_trend.value < current_price)
        elif self.market_state == "BEARISH":
            conditions.append(self.ema_trend.value > current_price)

        # 生成信号
        if all(conditions):
            return {
                'type': "BUY" if self.market_state == "BULLISH" else "SELL",
                'strength': self.trend_strength,
                'confidence': len(conditions) / 3,
            }

        return None

7.2.2 自定义移动平均

"""
自定义移动平均线指标
"""

from decimal import Decimal
import math


class HullMovingAverage(Indicator):
    """赫尔移动平均线 (HMA)"""

    def __init__(self, period: int = 20):
        """
        初始化HMA

        Parameters
        ----------
        period : int
            周期
        """
        super().__init__(params={'period': period}, name="HMA")
        self.period = period
        self.half_period = int(period / 2)
        self.sqrt_period = int(math.sqrt(period))

        # 内部WMA指标
        self.wma_full = WMA(period=period)
        self.wma_half = WMA(period=self.half_period)
        self.wma_diff = WMA(period=self.sqrt_period)

        self._value = 0.0

    @property
    def value(self) -> float:
        """获取HMA值"""
        return self._value

    def handle_bar(self, bar: Bar):
        """处理K线数据"""
        self.update_raw(float(bar.close))

    def handle_quote_tick(self, quote):
        """处理报价数据"""
        mid_price = (float(quote.bid_price) + float(quote.ask_price)) / 2
        self.update_raw(mid_price)

    def handle_trade_tick(self, trade):
        """处理成交数据"""
        self.update_raw(float(trade.price))

    def reset(self):
        """重置指标"""
        self.wma_full.reset()
        self.wma_half.reset()
        self.wma_diff.reset()
        self._value = 0.0
        self._initialized = False
        self._has_inputs = False
        self._inputs = 0

    def _update(self, value: float):
        """更新HMA值"""
        # 更新内部WMA
        self.wma_full.update_raw(value)
        self.wma_half.update_raw(value)

        if self.wma_full.initialized and self.wma_half.initialized:
            # 计算2 * WMA(n/2) - WMA(n)
            diff_value = 2 * self.wma_half.value - self.wma_full.value
            self.wma_diff.update_raw(diff_value)

            if self.wma_diff.initialized:
                self._value = self.wma_diff.value
                self._initialized = True


class JurikMovingAverage(Indicator):
    """Jurik移动平均线 (JMA) - 平滑性优化的移动平均"""

    def __init__(
        self,
        period: int = 14,
        phase: float = 0.0,
        power: float = 2.0,
    ):
        """
        初始化JMA

        Parameters
        ----------
        period : int
            周期
        phase : float
            相位 (-100 to 100)
        power : float
            幂次
        """
        super().__init__(
            params={
                'period': period,
                'phase': phase,
                'power': power
            },
            name="JMA"
        )

        self.period = period
        self.phase = phase
        self.power = power

        # JMA参数
        self.phase_ratio = phase / 100.0 + 1.5
        self.beta = 0.45 * (period - 1) / (0.45 * (period - 1) + 2)
        self.alpha = self.beta

        # 内部变量
        self.prev_jma = 0.0
        self.prev_det0 = 0.0
        self.prev_det1 = 0.0

        self._value = 0.0
        self._initialized = False

    @property
    def value(self) -> float:
        """获取JMA值"""
        return self._value

    def handle_bar(self, bar: Bar):
        """处理K线数据"""
        self.update_raw(float(bar.close))

    def handle_quote_tick(self, quote):
        """处理报价数据"""
        mid_price = (float(quote.bid_price) + float(quote.ask_price)) / 2
        self.update_raw(mid_price)

    def handle_trade_tick(self, trade):
        """处理成交数据"""
        self.update_raw(float(trade.price))

    def reset(self):
        """重置指标"""
        self.prev_jma = 0.0
        self.prev_det0 = 0.0
        self.prev_det1 = 0.0
        self._value = 0.0
        self._initialized = False
        self._has_inputs = False
        self._inputs = 0

    def _update(self, value: float):
        """更新JMA值"""
        if not self.has_inputs:
            # 第一次输入
            self.prev_jma = value
            self.prev_det0 = 0.0
            self.prev_det1 = 0.0
            self._initialized = True
        else:
            # JMA计算
            det0 = (1 - self.alpha) * (value + value - self.prev_jma)
            det1 = self.alpha * self.prev_det0 + det0

            # 幂次变换
            if self.power != 1:
                abs_det1 = abs(det1) ** self.power
                if det1 < 0:
                    det1 = -abs_det1
                else:
                    det1 = abs_det1

            # 更新JMA
            self._value = self.prev_jma + self.phase_ratio * det1

            # 保存历史值
            self.prev_jma = self._value
            self.prev_det0 = det0
            self.prev_det1 = det1


# 使用自定义MA的策略
class CustomMAStrategy(Strategy):
    """使用自定义移动平均线的策略"""

    def __init__(self, bar_type: BarType):
        """初始化策略"""
        super().__init__()
        self.bar_type = bar_type

        # 自定义移动平均线
        self.hma_fast = HullMovingAverage(period=10)
        self.hma_slow = HullMovingAverage(period=30)
        self.jma_smooth = JurikMovingAverage(period=20, phase=0)

        # 状态跟踪
        self.hma_spread = 0.0
        self.jma_trend = 0.0

    def on_start(self):
        """策略启动"""
        self.log.info("自定义MA策略启动")
        self.subscribe_bars(self.bar_type)

    def on_bar(self, bar: Bar):
        """处理K线数据"""
        price = float(bar.close)

        # 更新指标
        self.hma_fast.update_raw(price)
        self.hma_slow.update_raw(price)
        self.jma_smooth.update_raw(price)

        # 等待初始化
        if not (self.hma_fast.initialized and
                self.hma_slow.initialized and
                self.jma_smooth.initialized):
            return

        # 计算指标差异
        self.hma_spread = (self.hma_fast.value - self.hma_slow.value) / self.hma_slow.value
        self.jma_trend = self.jma_smooth.value - self.jma_smooth.value_1

        # 生成信号
        self._check_signals(bar)

    def _check_signals(self, bar: Bar):
        """检查交易信号"""
        # HMA交叉信号
        hma_cross = (
            (self.hma_fast.value > self.hma_slow.value) and
            (self.hma_fast.value_1 <= self.hma_slow.value_1)
        ) or (
            (self.hma_fast.value < self.hma_slow.value) and
            (self.hma_fast.value_1 >= self.hma_slow.value_1)
        )

        # JMA趋势确认
        trend_confirm = abs(self.jma_trend) > 0.001

        if hma_cross and trend_confirm:
            signal = "BUY" if self.hma_fast.value > self.hma_slow.value else "SELL"
            self.log.info(
                f"信号: {signal}, HMA价差: {self.hma_spread:.4f}, "
                f"JMA趋势: {self.jma_trend:.4f}",
                color=LogColor.YELLOW
            )

7.3 动量指标

7.3.1 RSI 指标

"""
RSI (相对强弱指数) 使用示例
"""

from decimal import Decimal
from nautilus_trader.indicators.momentum import RSI


class RSIStrategy(Strategy):
    """RSI策略"""

    def __init__(
        self,
        bar_type: BarType,
        rsi_period: int = 14,
        oversold: float = 30,
        overbought: float = 70,
    ):
        """
        初始化RSI策略

        Parameters
        ----------
        bar_type : BarType
            K线类型
        rsi_period : int
            RSI周期
        oversold : float
            超卖线
        overbought : float
            超买线
        """
        super().__init__()
        self.bar_type = bar_type

        # 创建RSI指标
        self.rsi = RSI(period=rsi_period)

        # 参数
        self.oversold = oversold
        self.overbought = overbought

        # 状态
        self.last_rsi = 0.0
        self.divergences = []

    def on_start(self):
        """策略启动"""
        self.log.info(
            f"RSI策略启动 - 周期: {self.rsi.period}, "
            f"超卖: {self.oversold}, 超买: {self.overbought}"
        )
        self.subscribe_bars(self.bar_type)

    def on_bar(self, bar: Bar):
        """处理K线数据"""
        # 更新RSI
        self.rsi.update_raw(float(bar.close))

        if not self.rsi.initialized:
            return

        current_rsi = self.rsi.value
        price = float(bar.close)

        # 检查RSI信号
        self._check_rsi_signals(current_rsi, price)

        # 检查背离
        self._check_divergences(current_rsi, price)

        # 记录RSI值
        self.last_rsi = current_rsi

        # 定期输出
        if self.count_bars() % 20 == 0:
            self.log.info(
                f"RSI: {current_rsi:.2f}, 价格: {price:.2f}",
                color=LogColor.MAGENTA
            )

    def _check_rsi_signals(self, rsi: float, price: float):
        """检查RSI信号"""
        # 超卖区域反弹
        if self.last_rsi < self.oversold and rsi > self.oversold:
            self.log.info(
                f"RSI超卖反弹 - RSI: {rsi:.2f}",
                color=LogColor.GREEN
            )
            self._generate_buy_signal(rsi, price, "OVERSOLD_RECOVERY")

        # 超买区域回调
        elif self.last_rsi > self.overbought and rsi < self.overbought:
            self.log.info(
                f"RSI超买回调 - RSI: {rsi:.2f}",
                color=LogColor.RED
            )
            self._generate_sell_signal(rsi, price, "OVERBOUGHT_PULLBACK")

        # 中轴突破
        elif self.last_rsi < 50 and rsi > 50:
            self.log.info(
                f"RSI突破中轴 - RSI: {rsi:.2f}",
                color=LogColor.BLUE
            )
            self._generate_buy_signal(rsi, price, "CENTER_CROSS_UP")

        elif self.last_rsi > 50 and rsi < 50:
            self.log.info(
                f"RSI跌破中轴 - RSI: {rsi:.2f}",
                color=LogColor.ORANGE
            )
            self._generate_sell_signal(rsi, price, "CENTER_CROSS_DOWN")

    def _check_divergences(self, rsi: float, price: float):
        """检查背离"""
        # 需要维护价格和RSI的历史数据
        # 这里简化实现
        pass

    def _generate_buy_signal(self, rsi: float, price: float, reason: str):
        """生成买入信号"""
        signal = {
            'type': 'BUY',
            'rsi': rsi,
            'price': price,
            'reason': reason,
            'strength': self._calculate_signal_strength(rsi, 'BUY')
        }
        self.log.info(f"买入信号: {signal}", color=LogColor.GREEN)

    def _generate_sell_signal(self, rsi: float, price: float, reason: str):
        """生成卖出信号"""
        signal = {
            'type': 'SELL',
            'rsi': rsi,
            'price': price,
            'reason': reason,
            'strength': self._calculate_signal_strength(rsi, 'SELL')
        }
        self.log.info(f"卖出信号: {signal}", color=LogColor.RED)

    def _calculate_signal_strength(self, rsi: float, signal_type: str) -> float:
        """计算信号强度"""
        if signal_type == 'BUY':
            # 超卖程度越深，信号越强
            return min((50 - rsi) / 20, 1.0)
        else:
            # 超买程度越高，信号越强
            return min((rsi - 50) / 20, 1.0)


# RSI + MA 组合策略
class RSIWithMAStrategy(Strategy):
    """RSI与移动平均线组合策略"""

    def __init__(self, bar_type: BarType):
        """初始化策略"""
        super().__init__()
        self.bar_type = bar_type

        # 指标
        self.rsi = RSI(period=14)
        self.ema_fast = EMA(period=20)
        self.ema_slow = EMA(period=50)

        # 状态
        self.trend = "NEUTRAL"
        self.rsi_state = "NEUTRAL"

    def on_start(self):
        """策略启动"""
        self.log.info("RSI+MA组合策略启动")
        self.subscribe_bars(self.bar_type)

    def on_bar(self, bar: Bar):
        """处理K线数据"""
        price = float(bar.close)

        # 更新指标
        self.rsi.update_raw(price)
        self.ema_fast.update_raw(price)
        self.ema_slow.update_raw(price)

        if not (self.rsi.initialized and
                self.ema_fast.initialized and
                self.ema_slow.initialized):
            return

        # 判断趋势
        self._determine_trend()

        # 判断RSI状态
        self._determine_rsi_state()

        # 生成信号
        self._generate_combined_signal(price)

    def _determine_trend(self):
        """判断趋势"""
        if self.ema_fast.value > self.ema_slow.value:
            self.trend = "BULLISH"
        else:
            self.trend = "BEARISH"

    def _determine_rsi_state(self):
        """判断RSI状态"""
        if self.rsi.value < 30:
            self.rsi_state = "OVERSOLD"
        elif self.rsi.value > 70:
            self.rsi_state = "OVERBOUGHT"
        else:
            self.rsi_state = "NEUTRAL"

    def _generate_combined_signal(self, price: float):
        """生成组合信号"""
        # 牛市 + RSI超卖 = 买入
        if (self.trend == "BULLISH" and
            self.rsi_state == "OVERSOLD"):
            self.log.info(
                f"组合买入信号 - 趋势: {self.trend}, RSI: {self.rsi.value:.1f}",
                color=LogColor.GREEN
            )

        # 熊市 + RSI超买 = 卖出
        elif (self.trend == "BEARISH" and
              self.rsi_state == "OVERBOUGHT"):
            self.log.info(
                f"组合卖出信号 - 趋势: {self.trend}, RSI: {self.rsi.value:.1f}",
                color=LogColor.RED
            )

7.3.2 MACD 指标

"""
MACD (移动平均收敛散度) 使用示例
"""

from nautilus_trader.indicators.momentum import MACD


class MACDStrategy(Strategy):
    """MACD策略"""

    def __init__(
        self,
        bar_type: BarType,
        fast_period: int = 12,
        slow_period: int = 26,
        signal_period: int = 9,
    ):
        """
        初始化MACD策略

        Parameters
        ----------
        bar_type : BarType
            K线类型
        fast_period : int
            快线周期
        slow_period : int
            慢线周期
        signal_period : int
            信号线周期
        """
        super().__init__()
        self.bar_type = bar_type

        # 创建MACD指标
        self.macd = MACD(
            fast_period=fast_period,
            slow_period=slow_period,
            signal_period=signal_period,
        )

        # 状态
        self.last_signal = 0  # MACD柱状图的符号
        self.histogram_trend = []  # 柱状图趋势

    def on_start(self):
        """策略启动"""
        self.log.info(
            f"MACD策略启动 - "
            f"快线: {self.macd.fast_period}, "
            f"慢线: {self.macd.slow_period}, "
            f"信号: {self.macd.signal_period}"
        )
        self.subscribe_bars(self.bar_type)

    def on_bar(self, bar: Bar):
        """处理K线数据"""
        # 更新MACD
        self.macd.update_raw(float(bar.close))

        if not self.macd.initialized:
            return

        # 获取MACD值
        macd_line = self.macd.value
        signal_line = self.macd.signal
        histogram = self.macd.histogram

        # 检查信号
        self._check_macd_signals(macd_line, signal_line, histogram, bar)

        # 记录数值
        if self.count_bars() % 10 == 0:
            self.log.info(
                f"MACD - 线: {macd_line:.4f}, "
                f"信号: {signal_line:.4f}, "
                f"柱: {histogram:.4f}",
                color=LogColor.CYAN
            )

    def _check_macd_signals(
        self,
        macd_line: float,
        signal_line: float,
        histogram: float,
        bar: Bar
    ):
        """检查MACD信号"""
        # 记录柱状图趋势
        self.histogram_trend.append(histogram)
        if len(self.histogram_trend) > 3:
            self.histogram_trend.pop(0)

        # 1. MACD线与信号线交叉
        if (macd_line > signal_line and self.last_signal <= 0):
            self._on_macd_bull_cross(bar)
        elif (macd_line < signal_line and self.last_signal >= 0):
            self._on_macd_bear_cross(bar)

        # 2. 柱状图背离
        if len(self.histogram_trend) >= 3:
            self._check_histogram_divergence(bar)

        # 3. 零轴突破
        if self.last_signal * histogram < 0:  # 符号变化
            if histogram > 0:
                self._on_zero_cross_up(bar)
            else:
                self._on_zero_cross_down(bar)

        self.last_signal = histogram

    def _on_macd_bull_cross(self, bar: Bar):
        """MACD牛市交叉"""
        self.log.info(
            f"MACD牛市交叉 - 价格: {bar.close:.2f}",
            color=LogColor.GREEN
        )

    def _on_macd_bear_cross(self, bar: Bar):
        """MACD熊市交叉"""
        self.log.info(
            f"MACD熊市交叉 - 价格: {bar.close:.2f}",
            color=LogColor.RED
        )

    def _on_zero_cross_up(self, bar: Bar):
        """MACD上穿零轴"""
        self.log.info(
            f"MACD上穿零轴 - 价格: {bar.close:.2f}",
            color=LogColor.BLUE
        )

    def _on_zero_cross_down(self, bar: Bar):
        """MACD下穿零轴"""
        self.log.info(
            f"MACD下穿零轴 - 价格: {bar.close:.2f}",
            color=LogColor.ORANGE
        )

    def _check_histogram_divergence(self, bar: Bar):
        """检查柱状图背离"""
        # 简化的背离检测
        h0, h1, h2 = self.histogram_trend[-3:]
        current_price = float(bar.close)

        # 看涨背离：价格新低但柱状图谷底抬高
        if (h2 < h1 < h0 and h0 < 0):  # 上升的谷底
            # 需要结合价格数据进行完整实现
            pass

        # 看跌背离：价格新高但柱状图峰顶降低
        elif (h2 > h1 > h0 and h0 > 0):  # 下降的峰顶
            # 需要结合价格数据进行完整实现
            pass

7.4 趋势和波动率指标

7.4.1 布林带 (Bollinger Bands)

"""
布林带使用示例
"""

from nautilus_trader.indicators.volatility import BollingerBands
from nautilus_trader.indicators.volatility import KeltnerChannels
from nautilus_trader.indicators.volatility import DonchianChannels


class BollingerBandsStrategy(Strategy):
    """布林带策略"""

    def __init__(
        self,
        bar_type: BarType,
        period: int = 20,
        std_dev: float = 2.0,
    ):
        """
        初始化布林带策略

        Parameters
        ----------
        bar_type : BarType
            K线类型
        period : int
            均线周期
        std_dev : float
            标准差倍数
        """
        super().__init__()
        self.bar_type = bar_type

        # 布林带指标
        self.bbands = BollingerBands(
            period=period,
            std_dev=std_dev,
        )

        # 辅助指标
        self.keltner = KeltnerChannels(
            period=period,
            multiplier=2.0,
        )
        self.donchian = DonchianChannels(period=period)

        # 状态
        self.price_position = "MIDDLE"  # UPPER, MIDDLE, LOWER
        self.band_width = 0.0
        self.squeeze_detected = False

    def on_start(self):
        """策略启动"""
        self.log.info(
            f"布林带策略启动 - "
            f"周期: {self.bbands.period}, "
            f"标准差: {self.bbands.std_dev}"
        )
        self.subscribe_bars(self.bar_type)

    def on_bar(self, bar: Bar):
        """处理K线数据"""
        price = float(bar.close)
        high = float(bar.high)
        low = float(bar.low)

        # 更新指标
        self.bbands.update_raw(price)
        self.keltner.update_raw(price)
        self.donchian.update_high(high)
        self.donchian.update_low(low)

        if not (self.bbands.initialized and
                self.keltner.initialized and
                self.donchian.initialized):
            return

        # 获取布林带值
        bb_upper = self.bbands.upper
        bb_middle = self.bbands.middle
        bb_lower = self.bbands.lower

        # 计算价格位置
        if price > bb_upper:
            self.price_position = "ABOVE_UPPER"
        elif price < bb_lower:
            self.price_position = "BELOW_LOWER"
        elif price > bb_middle:
            self.price_position = "UPPER_HALF"
        else:
            self.price_position = "LOWER_HALF"

        # 计算带宽
        self.band_width = (bb_upper - bb_lower) / bb_middle

        # 检测挤压
        self._detect_squeeze()

        # 生成信号
        self._check_signals(price, bar)

        # 记录信息
        if self.count_bars() % 10 == 0:
            self.log.info(
                f"BB - 上轨: {bb_upper:.2f}, "
                f"中轨: {bb_middle:.2f}, "
                f"下轨: {bb_lower:.2f}, "
                f"价格: {price:.2f}",
                color=LogColor.MAGENTA
            )

    def _detect_squeeze(self):
        """检测布林带挤压"""
        # 布林带宽度小于某个阈值
        min_width = 0.02  # 2%

        if self.band_width < min_width:
            if not self.squeeze_detected:
                self.squeeze_detected = True
                self.log.warning(
                    f"布林带挤压检测 - 带宽: {self.band_width:.4f}",
                    color=LogColor.YELLOW
                )
        else:
            if self.squeeze_detected:
                self.squeeze_detected = False
                self.log.info("布林带挤压释放", color=LogColor.CYAN)

    def _check_signals(self, price: float, bar: Bar):
        """检查交易信号"""
        bb_upper = self.bbands.upper
        bb_lower = self.bbands.lower

        # 回归交易（均值回归）
        if self.price_position == "ABOVE_UPPER":
            self._generate_sell_signal(
                price=price,
                reason="UPPER_BAND_REJECTION",
                strength=self._calculate_strength("mean_reversion")
            )
        elif self.price_position == "BELOW_LOWER":
            self._generate_buy_signal(
                price=price,
                reason="LOWER_BAND_SUPPORT",
                strength=self._calculate_strength("mean_reversion")
            )

        # 突破交易
        elif self.price_position == "UPPER_HALF":
            # 检查是否突破
            if bar.close > bar.open and self.band_width > 0.04:
                self._generate_buy_signal(
                    price=price,
                    reason="UPPER_BAND_BREAKOUT",
                    strength=self._calculate_strength("breakout")
                )
        elif self.price_position == "LOWER_HALF":
            # 检查是否跌破
            if bar.close < bar.open and self.band_width > 0.04:
                self._generate_sell_signal(
                    price=price,
                    reason="LOWER_BAND_BREAKOUT",
                    strength=self._calculate_strength("breakout")
                )

    def _calculate_strength(self, signal_type: str) -> float:
        """计算信号强度"""
        if signal_type == "mean_reversion":
            # 带宽越窄，回归信号越强
            return max(1.0 - self.band_width * 10, 0.3)
        elif signal_type == "breakout":
            # 带宽越宽，突破信号越强
            return min(self.band_width * 15, 1.0)
        return 0.5

    def _generate_buy_signal(self, price: float, reason: str, strength: float):
        """生成买入信号"""
        self.log.info(
            f"买入信号 - 价格: {price:.2f}, "
            f"原因: {reason}, 强度: {strength:.2f}",
            color=LogColor.GREEN
        )

    def _generate_sell_signal(self, price: float, reason: str, strength: float):
        """生成卖出信号"""
        self.log.info(
            f"卖出信号 - 价格: {price:.2f}, "
            f"原因: {reason}, 强度: {strength:.2f}",
            color=LogColor.RED
        )

7.5 自定义指标开发

7.5.1 创建复合指标

"""
自定义复合指标示例
"""

from typing import Optional
import numpy as np


class AdaptiveRSI(Indicator):
    """自适应RSI指标"""

    def __init__(
        self,
        base_period: int = 14,
        min_period: int = 5,
        max_period: int = 30,
        sensitivity: float = 0.02,
    ):
        """
        初始化自适应RSI

        Parameters
        ----------
        base_period : int
            基础周期
        min_period : int
            最小周期
        max_period : int
            最大周期
        sensitivity : float
            敏感度
        """
        super().__init__(
            params={
                'base_period': base_period,
                'min_period': min_period,
                'max_period': max_period,
                'sensitivity': sensitivity
            },
            name="AdaptiveRSI"
        )

        self.base_period = base_period
        self.min_period = min_period
        self.max_period = max_period
        self.sensitivity = sensitivity

        # 内部RSI
        self.rsi = RSI(period=base_period)

        # 波动率计算
        self.price_changes = []
        self.volatility = 0.0

        # 自适应周期
        self.current_period = base_period

        self._value = 0.0

    @property
    def value(self) -> float:
        """获取指标值"""
        return self._value

    @property
    def adaptive_period(self) -> int:
        """获取自适应周期"""
        return self.current_period

    def handle_bar(self, bar: Bar):
        """处理K线数据"""
        self.update_raw(float(bar.close))

    def handle_quote_tick(self, quote):
        """处理报价数据"""
        mid_price = (float(quote.bid_price) + float(quote.ask_price)) / 2
        self.update_raw(mid_price)

    def handle_trade_tick(self, trade):
        """处理成交数据"""
        self.update_raw(float(trade.price))

    def reset(self):
        """重置指标"""
        self.rsi.reset()
        self.price_changes.clear()
        self.volatility = 0.0
        self.current_period = self.base_period
        self._value = 0.0
        self._initialized = False
        self._has_inputs = False
        self._inputs = 0

    def _update(self, value: float):
        """更新指标值"""
        # 计算价格变化
        if self.has_inputs:
            price_change = abs(value - self.price_changes[-1])
            self.price_changes.append(price_change)

            # 保留足够的历史数据
            if len(self.price_changes) > self.max_period * 2:
                self.price_changes.pop(0)

            # 计算波动率
            if len(self.price_changes) >= self.base_period:
                self.volatility = np.std(self.price_changes[-self.base_period:])

                # 自适应调整周期
                self._adjust_period()

        else:
            self.price_changes.append(value)

        # 更新内部RSI
        self.rsi.update_raw(value)

        # 调整RSI周期
        if self.rsi.period != self.current_period:
            self.rsi = RSI(period=self.current_period)
            # 重新计算历史数据
            for price in self.price_changes[-self.current_period:]:
                self.rsi.update_raw(price)

        self._value = self.rsi.value

        if self.rsi.initialized:
            self._initialized = True

    def _adjust_period(self):
        """调整RSI周期"""
        # 基于波动率调整周期
        # 高波动率 -> 短周期
        # 低波动率 -> 长周期

        volatility_ratio = self.volatility / (np.mean(self.price_changes) + 1e-8)

        # 计算新周期
        period_adjustment = int(volatility_ratio * self.sensitivity * self.base_period)
        new_period = self.base_period - period_adjustment

        # 限制周期范围
        self.current_period = max(self.min_period, min(self.max_period, new_period))


class MultiTimeframeRSI(Indicator):
    """多时间框架RSI指标"""

    def __init__(
        self,
        short_period: int = 7,
        medium_period: int = 14,
        long_period: int = 28,
    ):
        """
        初始化多时间框架RSI

        Parameters
        ----------
        short_period : int
            短周期
        medium_period : int
            中周期
        long_period : int
            长周期
        """
        super().__init__(
            params={
                'short_period': short_period,
                'medium_period': medium_period,
                'long_period': long_period
            },
            name="MultiTimeframeRSI"
        )

        # 三个不同周期的RSI
        self.rsi_short = RSI(period=short_period)
        self.rsi_medium = RSI(period=medium_period)
        self.rsi_long = RSI(period=long_period)

        # 综合信号
        self._value = 50.0
        self.signal_strength = 0.0

    @property
    def value(self) -> float:
        """获取综合RSI值"""
        return self._value

    @property
    def strength(self) -> float:
        """获取信号强度"""
        return self.signal_strength

    @property
    def rsi_values(self) -> dict:
        """获取各周期RSI值"""
        return {
            'short': self.rsi_short.value if self.rsi_short.initialized else None,
            'medium': self.rsi_medium.value if self.rsi_medium.initialized else None,
            'long': self.rsi_long.value if self.rsi_long.initialized else None,
        }

    def handle_bar(self, bar: Bar):
        """处理K线数据"""
        self.update_raw(float(bar.close))

    def handle_quote_tick(self, quote):
        """处理报价数据"""
        mid_price = (float(quote.bid_price) + float(quote.ask_price)) / 2
        self.update_raw(mid_price)

    def handle_trade_tick(self, trade):
        """处理成交数据"""
        self.update_raw(float(trade.price))

    def reset(self):
        """重置指标"""
        self.rsi_short.reset()
        self.rsi_medium.reset()
        self.rsi_long.reset()
        self._value = 50.0
        self.signal_strength = 0.0
        self._initialized = False
        self._has_inputs = False
        self._inputs = 0

    def _update(self, value: float):
        """更新指标值"""
        # 更新所有RSI
        self.rsi_short.update_raw(value)
        self.rsi_medium.update_raw(value)
        self.rsi_long.update_raw(value)

        # 等待所有RSI初始化
        if not (self.rsi_short.initialized and
                self.rsi_medium.initialized and
                self.rsi_long.initialized):
            return

        # 计算加权平均
        weights = [0.5, 0.3, 0.2]  # 短期权重最大
        rsi_values = [
            self.rsi_short.value,
            self.rsi_medium.value,
            self.rsi_long.value
        ]

        self._value = sum(w * r for w, r in zip(weights, rsi_values))

        # 计算信号强度（基于RSI的一致性）
        rsi_mean = np.mean(rsi_values)
        rsi_std = np.std(rsi_values)
        self.signal_strength = max(0, 1 - rsi_std / 50)  # 标准化到0-1

        self._initialized = True

7.6 性能优化

7.6.1 批量处理和缓存

"""
指标性能优化
"""

from typing import List, Tuple
import numpy as np
from functools import lru_cache


class OptimizedIndicator(Indicator):
    """优化后的指标基类"""

    def __init__(self, batch_size: int = 100):
        """
        初始化优化指标

        Parameters
        ----------
        batch_size : int
            批处理大小
        """
        super().__init__()
        self.batch_size = batch_size
        self.buffer = []
        self._cache_enabled = True
        self._update_count = 0

    @lru_cache(maxsize=1024)
    def _cached_calculation(self, value: float) -> float:
        """缓存的计算（适用于纯函数）"""
        # 子类实现具体的计算逻辑
        return value

    def update_batch(self, values: List[float]):
        """
        批量更新

        Parameters
        ----------
        values : List[float]
            值列表
        """
        self.buffer.extend(values)

        # 达到批处理大小时进行批量处理
        if len(self.buffer) >= self.batch_size:
            self._process_batch()
            self.buffer.clear()

    def _process_batch(self):
        """处理批量数据"""
        # 使用numpy进行向量化计算
        np_values = np.array(self.buffer)

        # 子类实现具体的批量处理逻辑
        self._batch_update(np_values)

    def _batch_update(self, values: np.ndarray):
        """批量更新（子类实现）"""
        pass


class OptimizedEMA(OptimizedIndicator):
    """优化的EMA指标"""

    def __init__(
        self,
        period: int = 20,
        batch_size: int = 100,
        cache_size: int = 1024,
    ):
        """
        初始化优化的EMA

        Parameters
        ----------
        period : int
            EMA周期
        batch_size : int
            批处理大小
        cache_size : int
            缓存大小
        """
        super().__init__(batch_size= batch_size)
        self.period = period
        self.alpha = 2.0 / (period + 1)
        self._value = 0.0
        self._initialized = False

        # 启用缓存
        self._cache_enabled = cache_size > 0
        if self._cache_enabled:
            self._cached_calculation = lru_cache(maxsize=cache_size)(self._calculate_ema)

    @property
    def value(self) -> float:
        """获取EMA值"""
        return self._value

    def handle_bar(self, bar: Bar):
        """处理K线数据"""
        self.update_raw(float(bar.close))

    def handle_quote_tick(self, quote):
        """处理报价数据"""
        mid_price = (float(quote.bid_price) + float(quote.ask_price)) / 2
        self.update_raw(mid_price)

    def handle_trade_tick(self, trade):
        """处理成交数据"""
        self.update_raw(float(trade.price))

    def reset(self):
        """重置指标"""
        self._value = 0.0
        self._initialized = False
        self.buffer.clear()
        self._update_count = 0

        # 清除缓存
        if hasattr(self._cached_calculation, 'cache_clear'):
            self._cached_calculation.cache_clear()

    def _update(self, value: float):
        """更新EMA值"""
        if self._cache_enabled:
            # 使用缓存的计算
            self._value = self._cached_calculation(value)
        else:
            # 直接计算
            self._value = self._calculate_ema(value)

        if not self._initialized:
            self._initialized = True

        self._update_count += 1

    def _calculate_ema(self, value: float) -> float:
        """计算EMA值"""
        if self._update_count == 0:
            # 第一个值
            return value
        else:
            # EMA公式
            return (value * self.alpha) + (self._value * (1 - self.alpha))

    def _batch_update(self, values: np.ndarray):
        """批量更新EMA"""
        # 使用numpy进行向量化计算
        for value in values:
            self._update(float(value))


class IndicatorManager:
    """指标管理器"""

    def __init__(self):
        """初始化管理器"""
        self.indicators = {}
        self.update_queue = []

    def add_indicator(self, name: str, indicator: Indicator):
        """
        添加指标

        Parameters
        ----------
        name : str
            指标名称
        indicator : Indicator
            指标实例
        """
        self.indicators[name] = indicator

    def update_all(self, value: float):
        """
        更新所有指标

        Parameters
        ----------
        value : float
            更新值
        """
        for indicator in self.indicators.values():
            indicator.update_raw(value)

    def update_indicators(self, updates: List[Tuple[str, float]]):
        """
        更新指定指标

        Parameters
        ----------
        updates : List[Tuple[str, float]]
            (指标名称, 值) 列表
        """
        for name, value in updates:
            if name in self.indicators:
                self.indicators[name].update_raw(value)

    def get_values(self) -> dict:
        """获取所有指标的值"""
        values = {}
        for name, indicator in self.indicators.items():
            if indicator.initialized:
                values[name] = indicator.value
        return values

    def reset_all(self):
        """重置所有指标"""
        for indicator in self.indicators.values():
            indicator.reset()

    def get_initialized_indicators(self) -> List[str]:
        """获取已初始化的指标列表"""
        return [
            name for name, indicator in self.indicators.items()
            if indicator.initialized
        ]


# 性能分析工具
class IndicatorProfiler:
    """指标性能分析器"""

    def __init__(self):
        """初始化分析器"""
        self.timings = {}
        self.call_counts = {}

    def profile_update(self, indicator_name: str, func, *args, **kwargs):
        """
        分析指标更新

        Parameters
        ----------
        indicator_name : str
            指标名称
        func : Callable
            更新函数
        *args, **kwargs : Any
            函数参数
        """
        import time

        start_time = time.perf_counter()
        result = func(*args, **kwargs)
        end_time = time.perf_counter()

        elapsed = (end_time - start_time) * 1000  # 转换为毫秒

        # 记录时间
        if indicator_name not in self.timings:
            self.timings[indicator_name] = []
        self.timings[indicator_name].append(elapsed)

        # 记录调用次数
        self.call_counts[indicator_name] = self.call_counts.get(indicator_name, 0) + 1

        return result

    def get_stats(self) -> dict:
        """获取性能统计"""
        stats = {}

        for name, times in self.timings.items():
            count = self.call_counts.get(name, 0)
            if count > 0:
                stats[name] = {
                    'calls': count,
                    'total_ms': sum(times),
                    'avg_ms': np.mean(times),
                    'max_ms': max(times),
                    'min_ms': min(times),
                    'std_ms': np.std(times),
                }

        return stats

    def find_slow_indicators(self, threshold_ms: float = 1.0) -> List[str]:
        """找出慢速指标"""
        slow_indicators = []
        stats = self.get_stats()

        for name, stat in stats.items():
            if stat['avg_ms'] > threshold_ms:
                slow_indicators.append(name)

        return slow_indicators


# 使用示例
def main():
    """性能优化示例"""
    from time import time

    # 创建指标管理器
    manager = IndicatorManager()

    # 添加指标
    manager.add_indicator("EMA_20", OptimizedEMA(period=20, batch_size=100))
    manager.add_indicator("RSI_14", RSI(period=14))
    manager.add_indicator("AdaptiveRSI", AdaptiveRSI(base_period=14))

    # 创建性能分析器
    profiler = IndicatorProfiler()

    # 模拟数据更新
    data = [100 + i * 0.1 + (i % 10) for i in range(1000)]

    # 测试性能
    start_time = time()
    for i, value in enumerate(data):
        # 使用分析器包装更新
        for name in ["EMA_20", "RSI_14", "AdaptiveRSI"]:
            indicator = manager.indicators[name]
            profiler.profile_update(name, indicator.update_raw, value)

    end_time = time()

    # 输出结果
    print(f"更新 {len(data)} 个数据点耗时: {end_time - start_time:.3f} 秒")

    # 性能统计
    stats = profiler.get_stats()
    print("\n指标性能统计:")
    for name, stat in stats.items():
        print(f"{name}:")
        print(f"  调用次数: {stat['calls']}")
        print(f"  平均耗时: {stat['avg_ms']:.3f} ms")
        print(f"  最大耗时: {stat['max_ms']:.3f} ms")

    # 找出慢速指标
    slow_indicators = profiler.find_slow_indicators()
    if slow_indicators:
        print(f"\n慢速指标 (>1ms): {slow_indicators}")
    else:
        print("\n所有指标性能良好")


if __name__ == "__main__":
    main()

7.7 下一步

在本章中，我们学习了：

NautilusTrader 技术指标系统的架构
内置指标的使用方法
移动平均线、动量、趋势指标的应用
自定义指标开发
性能优化技巧

在下一章中，我们将学习：

订单管理系统
各种订单类型的使用
订单状态管理
执行算法
风险控制

7.8 总结

关键要点

NautilusTrader 提供了丰富的内置指标库
指标系统采用统一的接口设计
支持自定义指标和复合指标
提供性能优化工具和批量处理能力

最佳实践

合理选择指标周期，避免过度优化
使用指标管理器统一管理多个指标
定期分析指标性能，找出瓶颈
结合多个指标进行信号确认

7.9 参考资料

NautilusTrader第6章：时间管理和事件处理

Henry Lin — Thu, 09 Jul 2026 01:03:19 +0000

第6章：时间管理和事件处理

学习目标

通过本章学习，您将：

理解 NautilusTrader 的时间管理机制
掌握定时器的使用方法
学会事件驱动编程模式
了解消息总线和事件路由
实现复杂的调度逻辑

6.1 时间管理架构

6.1.1 时钟类型

NautilusTrader 提供了多种时钟实现，以适应不同的需求：

"""
时钟类型和使用场景
"""

from nautilus_trader.core.datetime import dt_to_unix_nanos
from nautilus_trader.test_kit.stubs.events import TestEventStubs
from datetime import datetime, timezone


# 1. LiveClock - 实时时钟（用于实盘交易）
from nautilus_trader.core.datetime import nanos_to_unix_nanos

class LiveClockExample:
    """实时时钟示例"""

    def __init__(self):
        from nautilus_trader.core.datetime import get_atomic_clock
        self.clock = get_atomic_clock()

    def get_current_time(self):
        """获取当前时间（纳秒）"""
        return self.clock.timestamp_ns()

    def wait_until(self, target_time_ns: int):
        """等待到指定时间"""
        import time
        current_time = self.clock.timestamp_ns()
        if current_time < target_time_ns:
            sleep_seconds = (target_time_ns - current_time) / 1_000_000_000
            time.sleep(sleep_seconds)

# 2. TestClock - 测试时钟（用于回测）
class TestClockExample:
    """测试时钟示例"""

    def __init__(self):
        from nautilus_trader.test_kit.stubs.events import TestClockStub
        # 创建测试时钟
        self.clock = TestClockStub()

        # 设置初始时间
        self.clock.set_time(dt_to_unix_nanos(datetime(2024, 1, 1)))

    def advance_time(self, seconds: int):
        """推进时间"""
        self.clock.advance_time(seconds * 1_000_000_000)

    def schedule_event(self, delay_ns: int, callback):
        """调度事件"""
        self.clock.set_time_ns_callback(
            target_time=self.clock.timestamp_ns() + delay_ns,
            callback=callback
        )


# 3. MonotonicClock - 单调时钟
class MonotonicClockExample:
    """单调时钟示例"""

    def __init__(self):
        from nautilus_trader.core.datetime import MonotonicClock
        self.clock = MonotonicClock()

    def measure_duration(self, func):
        """测量函数执行时间"""
        start = self.clock.timestamp_ns()
        func()
        end = self.clock.timestamp_ns()
        return end - start

6.1.2 时间格式和转换

"""
时间处理工具
"""

from datetime import datetime, timedelta, timezone
from decimal import Decimal
import pytz


class TimeUtils:
    """时间处理工具类"""

    @staticmethod
    def to_utc_naive(dt: datetime) -> datetime:
        """转换为UTC时间（无时区）"""
        if dt.tzinfo is None:
            return dt
        return dt.astimezone(timezone.utc).replace(tzinfo=None)

    @staticmethod
    def from_utc_naive(dt: datetime, tz_str: str = "UTC") -> datetime:
        """从UTC时间转换为指定时区"""
        tz = pytz.timezone(tz_str)
        return tz.localize(dt)

    @staticmethod
    def to_nanos(dt: datetime) -> int:
        """转换为纳秒时间戳"""
        return int(dt.timestamp() * 1_000_000_000)

    @staticmethod
    def from_nanos(nanos: int) -> datetime:
        """从纳秒时间戳转换"""
        return datetime.fromtimestamp(nanos / 1_000_000_000, tz=timezone.utc)

    @staticmethod
    def format_duration(nanos: int) -> str:
        """格式化时间间隔"""
        seconds = nanos / 1_000_000_000

        if seconds < 1:
            return f"{nanos / 1_000_000:.2f}ms"
        elif seconds < 60:
            return f"{seconds:.2f}s"
        elif seconds < 3600:
            minutes = seconds / 60
            return f"{minutes:.2f}min"
        else:
            hours = seconds / 3600
            return f"{hours:.2f}h"

    @staticmethod
    def get_trading_session_time(dt: datetime, timezone_str: str = "UTC") -> tuple:
        """
        获取交易时段

        Returns:
            tuple: (session_start, session_end, is_trading_time)
        """
        tz = pytz.timezone(timezone_str)
        local_dt = dt.astimezone(tz)

        # 标准交易时间：9:30 - 16:00
        session_start = local_dt.replace(
            hour=9, minute=30, second=0, microsecond=0
        )
        session_end = local_dt.replace(
            hour=16, minute=0, second=0, microsecond=0
        )

        is_trading = session_start <= local_dt <= session_end

        return session_start, session_end, is_trading


# 使用示例
def main():
    """时间处理示例"""

    utils = TimeUtils()

    # 当前时间
    now = datetime.now()
    print(f"本地时间: {now}")

    # 转换为UTC
    utc_time = utils.to_utc_naive(now)
    print(f"UTC时间: {utc_time}")

    # 转换为纳秒
    nanos = utils.to_nanos(utc_time)
    print(f"纳秒时间戳: {nanos}")

    # 转换回来
    dt_back = utils.from_nanos(nanos)
    print(f"转换回来: {dt_back}")

    # 交易时间检查
    session_start, session_end, is_trading = utils.get_trading_session_time(now)
    print(f"交易时段: {session_start} - {session_end}")
    print(f"当前是否交易时间: {is_trading}")


if __name__ == "__main__":
    main()

6.2 定时器系统

6.2.1 基础定时器

"""
定时器使用示例
"""

import asyncio
from datetime import datetime, timedelta
from decimal import Decimal

from nautilus_trader.core.datetime import dt_to_unix_nanos
from nautilus_trader.test_kit.stubs.events import TestClockStub
from nautilus_trader.trading.strategy import Strategy
from nautilus_trader.model.data import Bar
from nautilus_trader.model.events import OrderFilled


class TimerStrategy(Strategy):
    """使用定时器的策略示例"""

    def __init__(self):
        """初始化策略"""
        super().__init__()

        # 定时器ID管理
        self.timers = {
            'daily_reset': None,
            'hourly_update': None,
            'minute_check': None,
            'exit_check': None,
        }

        # 状态跟踪
        self.daily_high = Decimal('0')
        self.daily_low = Decimal('999999')
        self.hourly_volume = Decimal('0')
        self.last_bar_time = None
        self.positions_held_time = {}

    def on_start(self):
        """策略启动时设置定时器"""
        self.log.info("定时器策略启动")

        # 每日重置定时器（UTC 00:00）
        self._schedule_daily_reset()

        # 每小时更新
        self._schedule_hourly_update()

        # 每分钟检查
        self._schedule_minute_check()

        # 每30秒检查退出条件
        self._schedule_exit_check()

    def _schedule_daily_reset(self):
        """调度每日重置定时器"""
        # 计算下一个UTC 00:00
        now = self.clock.timestamp_ns()
        now_dt = datetime.fromtimestamp(now / 1_000_000_000, tz=timezone.utc)

        next_midnight = (now_dt + timedelta(days=1)).replace(
            hour=0, minute=0, second=0, microsecond=0
        )

        delay_ns = dt_to_unix_nanos(next_midnight) - now

        self.timers['daily_reset'] = self.clock.set_timer(
            name="daily_reset",
            interval_ns=delay_ns,
            callback=self._on_daily_reset,
        )

        self.log.info(f"每日重置定时器设置，下次执行: {next_midnight}")

    def _on_daily_reset(self):
        """每日重置回调"""
        self.log.info("执行每日重置")

        # 重置统计
        self.daily_high = Decimal('0')
        self.daily_low = Decimal('999999')
        self.hourly_volume = Decimal('0')

        # 清空持仓时间记录
        self.positions_held_time.clear()

        # 重新调度下次重置
        self._schedule_daily_reset()

    def _schedule_hourly_update(self):
        """调度每小时更新"""
        # 计算到下一个整小时的延迟
        now = self.clock.timestamp_ns()
        now_dt = datetime.fromtimestamp(now / 1_000_000_000, tz=timezone.utc)

        next_hour = (now_dt + timedelta(hours=1)).replace(
            minute=0, second=0, microsecond=0
        )

        delay_ns = dt_to_unix_nanos(next_hour) - now

        self.timers['hourly_update'] = self.clock.set_timer(
            name="hourly_update",
            interval_ns=delay_ns,
            callback=self._on_hourly_update,
        )

    def _on_hourly_update(self):
        """每小时更新回调"""
        self.log.info(f"每小时更新 - 当前小时成交量: {self.hourly_volume}")

        # 重置小时成交量
        self.hourly_volume = Decimal('0')

        # 重新调度
        self._schedule_hourly_update()

        # 这里可以执行其他小时级任务
        self._analyze_hourly_performance()

    def _schedule_minute_check(self):
        """调度每分钟检查"""
        self.timers['minute_check'] = self.clock.set_timer(
            name="minute_check",
            interval_ns=60 * 1_000_000_000,  # 1分钟
            callback=self._on_minute_check,
        )

    def _on_minute_check(self):
        """每分钟检查回调"""
        current_time = self.clock.timestamp_ns()

        # 检查是否有长时间未更新的数据
        if self.last_bar_time:
            time_diff = current_time - self.last_bar_time
            if time_diff > 2 * 60 * 1_000_000_000:  # 超过2分钟
                self.log.warning("数据更新异常 - 可能存在连接问题")

    def _schedule_exit_check(self):
        """调度退出检查"""
        self.timers['exit_check'] = self.clock.set_timer(
            name="exit_check",
            interval_ns=30 * 1_000_000_000,  # 30秒
            callback=self._on_exit_check,
        )

    def _on_exit_check(self):
        """退出条件检查"""
        # 检查持仓时间
        current_time = self.clock.timestamp_ns()

        for position_id, entry_time in list(self.positions_held_time.items()):
            hold_duration = current_time - entry_time
            hold_hours = hold_duration / (1_000_000_000 * 3600)

            # 持仓超过24小时强制平仓
            if hold_hours > 24:
                self.log.warning(
                    f"持仓时间过长 - 强制平仓: {position_id}, "
                    f"持仓时间: {hold_hours:.1f}小时"
                )
                self._force_close_position(position_id)

    def on_bar(self, bar: Bar):
        """处理K线数据"""
        # 更新最后K线时间
        self.last_bar_time = self.clock.timestamp_ns()

        # 更新日内高低点
        if bar.close > self.daily_high:
            self.daily_high = bar.close

        if bar.close < self.daily_low:
            self.daily_low = bar.close

        # 累计成交量
        self.hourly_volume += bar.volume

        # 检查交易信号
        self._check_trading_signal(bar)

    def on_order_filled(self, event: OrderFilled):
        """处理订单成交"""
        # 记录持仓时间
        self.positions_held_time[event.position_id] = self.clock.timestamp_ns()

    def _check_trading_signal(self, bar: Bar):
        """检查交易信号"""
        # 基于高低点突破的策略
        if bar.close > self.daily_high * Decimal('1.001'):  # 突破高点0.1%
            self._enter_long(bar)
        elif bar.close < self.daily_low * Decimal('0.999'):  # 跌破低点0.1%
            self._enter_short(bar)

    def _analyze_hourly_performance(self):
        """分析每小时表现"""
        # 这里可以实现每小时的性能分析
        pass

    def _force_close_position(self, position_id):
        """强制平仓"""
        # 实现强制平仓逻辑
        pass

    def on_stop(self):
        """策略停止时清理定时器"""
        self.log.info("停止定时器策略")

        # 取消所有定时器
        for timer_name, timer_id in self.timers.items():
            if timer_id:
                self.clock.cancel_timer(timer_id)
                self.log.info(f"取消定时器: {timer_name}")

6.2.2 高级定时器模式

"""
高级定时器模式
"""

from typing import Callable, Dict, List
from enum import Enum
import heapq


class TimerPriority(Enum):
    """定时器优先级"""
    CRITICAL = 1    # 关键任务（如风险检查）
    HIGH = 2        # 高优先级（如交易信号）
    NORMAL = 3      # 正常优先级（如统计更新）
    LOW = 4         # 低优先级（如清理任务）


class AdvancedTimerManager:
    """高级定时器管理器"""

    def __init__(self, clock):
        """初始化定时器管理器"""
        self.clock = clock
        self.timers = {}  # timer_id -> timer_info
        self.timer_heap = []  # 优先队列

    def schedule(
        self,
        name: str,
        callback: Callable,
        delay_ns: int,
        interval_ns: int = None,
        priority: TimerPriority = TimerPriority.NORMAL,
        metadata: Dict = None,
    ) -> str:
        """
        调度定时器

        Parameters
        ----------
        name : str
            定时器名称
        callback : Callable
            回调函数
        delay_ns : int
            延迟时间（纳秒）
        interval_ns : int
            重复间隔（纳秒），None表示单次
        priority : TimerPriority
            优先级
        metadata : Dict
            元数据

        Returns
        -------
        str
            定时器ID
        """
        current_time = self.clock.timestamp_ns()

        timer_info = {
            'id': f"{name}_{current_time}",
            'name': name,
            'callback': callback,
            'delay_ns': delay_ns,
            'interval_ns': interval_ns,
            'priority': priority,
            'metadata': metadata or {},
            'next_run': current_time + delay_ns,
            'run_count': 0,
            'active': True,
        }

        self.timers[timer_info['id']] = timer_info

        # 添加到优先队列
        heapq.heappush(
            self.timer_heap,
            (timer_info['next_run'], priority.value, timer_info['id'])
        )

        return timer_info['id']

    def cancel(self, timer_id: str):
        """取消定时器"""
        if timer_id in self.timers:
            self.timers[timer_id]['active'] = False
            del self.timers[timer_id]

    def process_pending(self):
        """处理待执行的定时器"""
        current_time = self.clock.timestamp_ns()

        while self.timer_heap:
            next_time, priority, timer_id = self.timer_heap[0]

            if next_time > current_time:
                break

            # 弹出定时器
            heapq.heappop(self.timer_heap)

            if timer_id not in self.timers:
                continue

            timer = self.timers[timer_id]
            if not timer['active']:
                continue

            # 执行回调
            try:
                timer['callback']()
                timer['run_count'] += 1
            except Exception as e:
                self.log.error(f"定时器执行错误: {timer_id}, 错误: {e}")

            # 如果是重复定时器，调度下次执行
            if timer['interval_ns']:
                timer['next_run'] = current_time + timer['interval_ns']
                heapq.heappush(
                    self.timer_heap,
                    (timer['next_run'], timer['priority'].value, timer_id)
                )
            else:
                # 单次定时器，删除
                del self.timers[timer_id]

    def get_stats(self) -> Dict:
        """获取统计信息"""
        total_timers = len(self.timers)

        # 按优先级统计
        priority_stats = {}
        for priority in TimerPriority:
            count = sum(
                1 for t in self.timers.values()
                if t['priority'] == priority and t['active']
            )
            priority_stats[priority.name] = count

        # 下次执行时间
        next_run_time = None
        if self.timer_heap:
            next_run_time = self.timer_heap[0][0]

        return {
            'total_active': total_timers,
            'by_priority': priority_stats,
            'next_run_ns': next_run_time,
        }


class ScheduledTaskStrategy(Strategy):
    """使用高级定时器的策略"""

    def __init__(self):
        """初始化策略"""
        super().__init__()

        # 任务调度器
        self.scheduler = AdvancedTimerManager(self.clock)

        # 统计数据
        self.stats = {
            'signals_detected': 0,
            'orders_submitted': 0,
            'risk_checks': 0,
            'data_updates': 0,
        }

    def on_start(self):
        """策略启动"""
        self.log.info("定时任务策略启动")

        # 调度各种任务

        # 1. 数据分析任务（每5分钟）
        self.scheduler.schedule(
            name="data_analysis",
            callback=self._analyze_market_data,
            delay_ns=5 * 60 * 1_000_000_000,  # 5分钟后开始
            interval_ns=5 * 60 * 1_000_000_000,  # 每5分钟
            priority=TimerPriority.NORMAL,
        )

        # 2. 风险检查任务（每分钟）
        self.scheduler.schedule(
            name="risk_check",
            callback=self._check_risk,
            delay_ns=60 * 1_000_000_000,  # 1分钟后开始
            interval_ns=60 * 1_000_000_000,  # 每分钟
            priority=TimerPriority.CRITICAL,
        )

        # 3. 信号扫描任务（每10秒）
        self.scheduler.schedule(
            name="signal_scan",
            callback=self._scan_signals,
            delay_ns=10 * 1_000_000_000,  # 10秒后开始
            interval_ns=10 * 1_000_000_000,  # 每10秒
            priority=TimerPriority.HIGH,
        )

        # 4. 性能统计任务（每小时）
        self.scheduler.schedule(
            name="performance_stats",
            callback=self._update_performance_stats,
            delay_ns=3600 * 1_000_000_000,  # 1小时后开始
            interval_ns=3600 * 1_000_000_000,  # 每小时
            priority=TimerPriority.LOW,
        )

        # 5. 清理任务（每天凌晨）
        self.scheduler.schedule(
            name="daily_cleanup",
            callback=self._daily_cleanup,
            delay_ns=self._get_time_to_next_midnight(),
            interval_ns=24 * 3600 * 1_000_000_000,  # 每天
            priority=TimerPriority.LOW,
        )

        # 启动定时器处理循环
        self.scheduler.schedule(
            name="timer_processor",
            callback=self._process_timers,
            delay_ns=100_000_000,  # 100ms后开始
            interval_ns=100_000_000,  # 每100ms
            priority=TimerPriority.CRITICAL,
        )

    def _process_timers(self):
        """处理待执行的定时器"""
        self.scheduler.process_pending()

    def _analyze_market_data(self):
        """分析市场数据"""
        self.stats['data_updates'] += 1
        self.log.debug(f"执行市场数据分析 - 更新次数: {self.stats['data_updates']}")

        # 获取最新数据
        # ... 分析逻辑 ...

    def _check_risk(self):
        """检查风险"""
        self.stats['risk_checks'] += 1
        self.log.debug(f"执行风险检查 - 检查次数: {self.stats['risk_checks']}")

        # 检查持仓风险
        # ... 风险逻辑 ...

    def _scan_signals(self):
        """扫描交易信号"""
        self.stats['signals_detected'] += 1
        self.log.debug(f"扫描交易信号 - 扫描次数: {self.stats['signals_detected']}")

        # 扫描市场信号
        # ... 信号逻辑 ...

    def _update_performance_stats(self):
        """更新性能统计"""
        self.log.info(
            f"性能统计 - "
            f"信号: {self.stats['signals_detected']}, "
            f"订单: {self.stats['orders_submitted']}, "
            f"风控: {self.stats['risk_checks']}"
        )

    def _daily_cleanup(self):
        """每日清理"""
        self.log.info("执行每日清理")

        # 清理过期数据
        # 重置统计
        # ... 清理逻辑 ...

    def _get_time_to_next_midnight(self) -> int:
        """获取到下一个午夜的时间"""
        now = datetime.now()
        next_midnight = (now + timedelta(days=1)).replace(
            hour=0, minute=0, second=0, microsecond=0
        )
        return int((next_midnight - now).total_seconds() * 1_000_000_000)

    def on_bar(self, bar: Bar):
        """处理K线数据"""
        # 定时器会自动处理，这里只处理实时数据
        pass

    def on_stop(self):
        """策略停止"""
        self.log.info("定时任务策略停止")

        # 显示最终统计
        stats = self.scheduler.get_stats()
        self.log.info(f"定时器统计: {stats}")
        self.log.info(f"任务统计: {self.stats}")

6.3 事件驱动架构

6.3.1 事件类型和继承

"""
事件系统架构
"""

from abc import ABC
from typing import Any, Dict, Optional
from datetime import datetime

from nautilus_trader.core.message import Event
from nautilus_trader.model.identifiers import TraderId, StrategyId


class CustomEvent(Event, ABC):
    """自定义事件基类"""

    def __init__(
        self,
        trader_id: TraderId,
        strategy_id: Optional[StrategyId] = None,
        event_id: Optional[str] = None,
        ts_init: Optional[int] = None,
        ts_event: Optional[int] = None,
        **kwargs: Any,
    ):
        """
        初始化自定义事件

        Parameters
        ----------
        trader_id : TraderId
            交易者ID
        strategy_id : StrategyId, optional
            策略ID
        event_id : str, optional
            事件ID
        ts_init : int, optional
            初始化时间戳
        ts_event : int, optional
            事件时间戳
        **kwargs : Any
            其他参数
        """
        super().__init__(
            trader_id=trader_id,
            strategy_id=strategy_id,
            event_id=event_id,
            ts_init=ts_init or self.clock.timestamp_ns(),
            ts_event=ts_event or self.clock.timestamp_ns(),
            **kwargs
        )

        self.metadata = kwargs


class SignalEvent(CustomEvent):
    """交易信号事件"""

    def __init__(
        self,
        trader_id: TraderId,
        strategy_id: StrategyId,
        instrument_id: str,
        signal_type: str,  # BUY, SELL, HOLD
        confidence: float,  # 0-1
        price: Optional[float] = None,
        quantity: Optional[float] = None,
        metadata: Optional[Dict] = None,
        **kwargs: Any,
    ):
        """
        初始化信号事件

        Parameters
        ----------
        instrument_id : str
            交易工具ID
        signal_type : str
            信号类型
        confidence : float
            置信度
        price : float, optional
            建议价格
        quantity : float, optional
            建议数量
        metadata : Dict, optional
            额外元数据
        """
        super().__init__(
            trader_id=trader_id,
            strategy_id=strategy_id,
            instrument_id=instrument_id,
            signal_type=signal_type,
            confidence=confidence,
            price=price,
            quantity=quantity,
            **(metadata or {})
        )

    def to_dict(self) -> Dict:
        """转换为字典"""
        return {
            'instrument_id': self.instrument_id,
            'signal_type': self.signal_type,
            'confidence': self.confidence,
            'price': self.price,
            'quantity': self.quantity,
            'timestamp': self.ts_event,
        }


class RiskEvent(CustomEvent):
    """风险事件"""

    def __init__(
        self,
        trader_id: TraderId,
        risk_type: str,  # POSITION_LIMIT, DRAWDOWN, CORRELATION
        risk_level: str,  # LOW, MEDIUM, HIGH, CRITICAL
        message: str,
        affected_positions: Optional[list] = None,
        **kwargs: Any,
    ):
        """
        初始化风险事件

        Parameters
        ----------
        risk_type : str
            风险类型
        risk_level : str
            风险级别
        message : str
            风险描述
        affected_positions : list, optional
            受影响的仓位
        """
        super().__init__(
            trader_id=trader_id,
            risk_type=risk_type,
            risk_level=risk_level,
            message=message,
            affected_positions=affected_positions or [],
            **kwargs
        )

    def requires_immediate_action(self) -> bool:
        """是否需要立即处理"""
        return self.risk_level in ['HIGH', 'CRITICAL']


class PerformanceEvent(CustomEvent):
    """性能统计事件"""

    def __init__(
        self,
        trader_id: TraderId,
        strategy_id: StrategyId,
        metric_type: str,  # PNL, DRAWDOWN, SHARPE, WIN_RATE
        metric_value: float,
        period: str,  # DAILY, WEEKLY, MONTHLY
        benchmark_value: Optional[float] = None,
        **kwargs: Any,
    ):
        """
        初始化性能事件

        Parameters
        ----------
        metric_type : str
            指标类型
        metric_value : float
            指标值
        period : str
            统计周期
        benchmark_value : float, optional
            基准值
        """
        super().__init__(
            trader_id=trader_id,
            strategy_id=strategy_id,
            metric_type=metric_type,
            metric_value=metric_value,
            period=period,
            benchmark_value=benchmark_value,
            **kwargs
        )

    def is_outperforming(self) -> bool:
        """是否跑赢基准"""
        if self.benchmark_value is None:
            return None
        return self.metric_value > self.benchmark_value


class SystemEvent(CustomEvent):
    """系统事件"""

    def __init__(
        self,
        trader_id: TraderId,
        system_type: str,  # CONNECTIVITY, DATA, EXECUTION
        severity: str,  # INFO, WARNING, ERROR, CRITICAL
        message: str,
        component: Optional[str] = None,
        **kwargs: Any,
    ):
        """
        初始化系统事件

        Parameters
        ----------
        system_type : str
            系统类型
        severity : str
            严重程度
        message : str
            事件消息
        component : str, optional
            组件名称
        """
        super().__init__(
            trader_id=trader_id,
            system_type=system_type,
            severity=severity,
            message=message,
            component=component,
            **kwargs
        )

    def requires_alert(self) -> bool:
        """是否需要告警"""
        return self.severity in ['ERROR', 'CRITICAL']

6.3.2 事件处理器和调度器

"""
事件处理系统
"""

from typing import Callable, Dict, List, Type
from collections import defaultdict
from concurrent.futures import ThreadPoolExecutor
import threading
import queue
import time


class EventBus:
    """事件总线"""

    def __init__(self, max_workers: int = 4):
        """
        初始化事件总线

        Parameters
        ----------
        max_workers : int
            最大工作线程数
        """
        self.handlers = defaultdict(list)
        self.executor = ThreadPoolExecutor(max_workers=max_workers)
        self.running = False
        self.event_queue = queue.Queue()
        self.processing_thread = None

    def subscribe(
        self,
        event_type: Type[Event],
        handler: Callable[[Event], None],
        priority: int = 0
    ):
        """
        订阅事件

        Parameters
        ----------
        event_type : Type[Event]
            事件类型
        handler : Callable
            处理函数
        priority : int
            优先级（数字越大优先级越高）
        """
        self.handlers[event_type].append((handler, priority))
        # 按优先级排序
        self.handlers[event_type].sort(key=lambda x: x[1], reverse=True)

    def unsubscribe(self, event_type: Type[Event], handler: Callable):
        """取消订阅"""
        handlers = self.handlers.get(event_type, [])
        self.handlers[event_type] = [
            (h, p) for h, p in handlers if h != handler
        ]

    def publish(self, event: Event):
        """
        发布事件（异步）

        Parameters
        ----------
        event : Event
            要发布的事件
        """
        self.event_queue.put(event)

    def publish_sync(self, event: Event):
        """
        发布事件（同步）

        Parameters
        ----------
        event : Event
            要发布的事件
        """
        self._process_event(event)

    def start(self):
        """启动事件处理"""
        if not self.running:
            self.running = True
            self.processing_thread = threading.Thread(
                target=self._process_loop,
                daemon=True
            )
            self.processing_thread.start()

    def stop(self):
        """停止事件处理"""
        self.running = False
        if self.processing_thread:
            self.processing_thread.join()

    def _process_loop(self):
        """事件处理循环"""
        while self.running:
            try:
                # 从队列获取事件
                event = self.event_queue.get(timeout=1.0)
                self._process_event(event)
                self.event_queue.task_done()
            except queue.Empty:
                continue
            except Exception as e:
                print(f"事件处理错误: {e}")

    def _process_event(self, event: Event):
        """处理单个事件"""
        event_type = type(event)
        handlers = self.handlers.get(event_type, [])

        # 并行处理所有订阅者
        futures = []
        for handler, priority in handlers:
            future = self.executor.submit(self._safe_handle, handler, event)
            futures.append(future)

        # 等待所有处理完成
        for future in futures:
            try:
                future.result(timeout=5.0)
            except Exception as e:
                print(f"处理器执行错误: {e}")

    def _safe_handle(self, handler: Callable, event: Event):
        """安全地调用处理器"""
        try:
            handler(event)
        except Exception as e:
            print(f"事件处理异常: {e}")

    def get_stats(self) -> Dict:
        """获取统计信息"""
        return {
            'queue_size': self.event_queue.qsize(),
            'subscriptions': {
                str(event_type): len(handlers)
                for event_type, handlers in self.handlers.items()
            },
            'running': self.running,
        }


class EventDrivenStrategy(Strategy):
    """事件驱动策略基类"""

    def __init__(self):
        """初始化策略"""
        super().__init__()

        # 事件总线
        self.event_bus = EventBus()

        # 事件处理器注册
        self._register_event_handlers()

        # 统计信息
        self.event_stats = defaultdict(int)

    def _register_event_handlers(self):
        """注册事件处理器"""
        # 注册自定义事件处理器
        self.event_bus.subscribe(SignalEvent, self._handle_signal_event)
        self.event_bus.subscribe(RiskEvent, self._handle_risk_event)
        self.event_bus.subscribe(PerformanceEvent, self._handle_performance_event)
        self.event_bus.subscribe(SystemEvent, self._handle_system_event)

    def on_start(self):
        """策略启动"""
        self.log.info("事件驱动策略启动")
        self.event_bus.start()

        # 发布启动事件
        self._publish_system_event("STARTUP", "INFO", "策略已启动")

    def on_stop(self):
        """策略停止"""
        self.log.info("事件驱动策略停止")
        self.event_bus.stop()

        # 发布停止事件
        self._publish_system_event("SHUTDOWN", "INFO", "策略已停止")

        # 显示事件统计
        self.log.info(f"事件统计: {dict(self.event_stats)}")

    def _handle_signal_event(self, event: SignalEvent):
        """处理信号事件"""
        self.event_stats['signals'] += 1

        self.log.info(
            f"收到信号事件: {event.signal_type} "
            f"(置信度: {event.confidence:.2f})"
        )

        # 根据信号执行交易
        if event.signal_type == 'BUY':
            self._execute_buy_signal(event)
        elif event.signal_type == 'SELL':
            self._execute_sell_signal(event)

    def _handle_risk_event(self, event: RiskEvent):
        """处理风险事件"""
        self.event_stats['risk_events'] += 1

        self.log.warning(
            f"风险事件: {event.risk_type} - {event.message}"
        )

        # 根据风险级别采取措施
        if event.requires_immediate_action():
            self._handle_critical_risk(event)
        else:
            self._handle_normal_risk(event)

    def _handle_performance_event(self, event: PerformanceEvent):
        """处理性能事件"""
        self.event_stats['performance_events'] += 1

        self.log.info(
            f"性能更新: {event.metric_type} = {event.metric_value} "
            f"({event.period})"
        )

        # 检查性能表现
        if event.is_outperforming() is False:
            self.log.warning("策略表现不佳，考虑调整参数")

    def _handle_system_event(self, event: SystemEvent):
        """处理系统事件"""
        self.event_stats['system_events'] += 1

        if event.requires_alert():
            self.log.error(f"系统告警: {event.message}")
        else:
            self.log.info(f"系统事件: {event.message}")

    def _execute_buy_signal(self, event: SignalEvent):
        """执行买入信号"""
        # 实现买入逻辑
        pass

    def _execute_sell_signal(self, event: SignalEvent):
        """执行卖出信号"""
        # 实现卖出逻辑
        pass

    def _handle_critical_risk(self, event: RiskEvent):
        """处理关键风险"""
        # 立即采取措施，如停止交易或平仓
        pass

    def _handle_normal_risk(self, event: RiskEvent):
        """处理普通风险"""
        # 记录日志，监控风险发展
        pass

    def _publish_signal_event(self, signal_type: str, confidence: float):
        """发布信号事件"""
        event = SignalEvent(
            trader_id=self.trader_id,
            strategy_id=self.id,
            instrument_id="BTCUSDT.BINANCE",
            signal_type=signal_type,
            confidence=confidence,
        )
        self.event_bus.publish(event)

    def _publish_risk_event(self, risk_type: str, risk_level: str, message: str):
        """发布风险事件"""
        event = RiskEvent(
            trader_id=self.trader_id,
            risk_type=risk_type,
            risk_level=risk_level,
            message=message,
        )
        self.event_bus.publish(event)

    def _publish_performance_event(self, metric_type: str, value: float, period: str):
        """发布性能事件"""
        event = PerformanceEvent(
            trader_id=self.trader_id,
            strategy_id=self.id,
            metric_type=metric_type,
            metric_value=value,
            period=period,
        )
        self.event_bus.publish(event)

    def _publish_system_event(self, system_type: str, severity: str, message: str):
        """发布系统事件"""
        event = SystemEvent(
            trader_id=self.trader_id,
            system_type=system_type,
            severity=severity,
            message=message,
        )
        self.event_bus.publish(event)


# 使用示例
def main():
    """主函数示例"""

    # 创建事件总线
    event_bus = EventBus()

    # 创建策略
    strategy = EventDrivenStrategy()

    # 模拟事件
    signal = SignalEvent(
        trader_id=TraderId("TRADER-001"),
        strategy_id=StrategyId("STRATEGY-001"),
        instrument_id="BTCUSDT.BINANCE",
        signal_type="BUY",
        confidence=0.85,
    )

    # 发布事件
    event_bus.publish(signal)

    # 运行一段时间后停止
    time.sleep(1)
    event_bus.stop()


if __name__ == "__main__":
    main()

6.4 实际应用示例

6.4.1 综合定时器和事件系统

"""
综合定时器和事件系统示例
"""

from nautilus_trader.trading.strategy import Strategy
from nautilus_trader.model.data import Bar
from nautilus_trader.model.events import OrderFilled
from nautilus_trader.model.objects import Money
from decimal import Decimal
from datetime import datetime, timedelta
import asyncio


class ComprehensiveStrategy(Strategy):
    """综合使用定时器和事件的策略"""

    def __init__(self):
        """初始化策略"""
        super().__init__()

        # 定时器管理
        self.timers = {
            'market_analysis': None,
            'risk_monitor': None,
            'performance_update': None,
            'news_check': None,
        }

        # 事件跟踪
        self.event_timeline = []
        self.signal_history = []

        # 状态变量
        self.market_state = {
            'trend': 'NEUTRAL',
            'volatility': 'NORMAL',
            'liquidity': 'GOOD',
        }

        # 性能指标
        self.performance_metrics = {
            'daily_pnl': Decimal('0'),
            'max_drawdown': Decimal('0'),
            'win_rate': Decimal('0'),
            'sharpe_ratio': Decimal('0'),
        }

    def on_start(self):
        """策略启动"""
        self.log.info("综合策略启动")

        # 设置定时器
        self._setup_timers()

        # 发布启动事件
        self._publish_system_event("STRATEGY_START", "INFO", "策略已启动")

    def _setup_timers(self):
        """设置所有定时器"""

        # 1. 市场分析（每5分钟）
        self.timers['market_analysis'] = self.clock.set_timer(
            name="market_analysis",
            interval_ns=5 * 60 * 1_000_000_000,
            callback=self._analyze_market,
        )

        # 2. 风险监控（每分钟）
        self.timers['risk_monitor'] = self.clock.set_timer(
            name="risk_monitor",
            interval_ns=60 * 1_000_000_000,
            callback=self._monitor_risk,
        )

        # 3. 性能更新（每小时）
        self.timers['performance_update'] = self.clock.set_timer(
            name="performance_update",
            interval_ns=3600 * 1_000_000_000,
            callback=self._update_performance,
        )

        # 4. 新闻检查（每15分钟）
        self.timers['news_check'] = self.clock.set_timer(
            name="news_check",
            interval_ns=15 * 60 * 1_000_000_000,
            callback=self._check_news,
        )

    def _analyze_market(self):
        """市场分析"""
        current_time = self.clock.timestamp_ns()

        # 获取最新市场数据
        # 这里应该从缓存或数据引擎获取
        # price = self.cache.get_last_price(...)

        # 模拟市场分析
        import random
        trend_change = random.choice(['UP', 'DOWN', 'NEUTRAL'])

        if trend_change != self.market_state['trend']:
            old_trend = self.market_state['trend']
            self.market_state['trend'] = trend_change

            # 发布趋势变化事件
            self._publish_market_event(
                "TREND_CHANGE",
                f"趋势从 {old_trend} 变为 {trend_change}",
                {
                    'old_trend': old_trend,
                    'new_trend': trend_change,
                }
            )

            # 生成交易信号
            if trend_change == 'UP':
                self._generate_signal('BUY', 0.75)
            elif trend_change == 'DOWN':
                self._generate_signal('SELL', 0.75)

    def _monitor_risk(self):
        """监控风险"""
        # 检查持仓风险
        total_exposure = self._calculate_total_exposure()
        max_allowed = Decimal('100000')  # 最大敞口10万

        if total_exposure > max_allowed:
            self._publish_risk_event(
                "EXPOSURE_LIMIT",
                "HIGH",
                f"总敞口超限: {total_exposure} > {max_allowed}",
            )

        # 检查回撤
        current_drawdown = self._calculate_drawdown()
        if current_drawdown > Decimal('0.1'):  # 10%
            self._publish_risk_event(
                "DRAWDOWN",
                "CRITICAL",
                f"回撤过大: {current_drawdown:.1%}",
            )

    def _update_performance(self):
        """更新性能指标"""
        # 计算各项指标
        self.performance_metrics['daily_pnl'] = self._calculate_daily_pnl()
        self.performance_metrics['max_drawdown'] = self._calculate_drawdown()
        self.performance_metrics['win_rate'] = self._calculate_win_rate()
        self.performance_metrics['sharpe_ratio'] = self._calculate_sharpe_ratio()

        # 发布性能事件
        for metric, value in self.performance_metrics.items():
            self._publish_performance_event(
                metric.upper(),
                float(value),
                "HOURLY"
            )

    def _check_news(self):
        """检查新闻（模拟）"""
        # 模拟新闻事件
        import random
        news_impact = random.choices(
            ['POSITIVE', 'NEGATIVE', 'NEUTRAL'],
            weights=[0.2, 0.1, 0.7]
        )[0]

        if news_impact != 'NEUTRAL':
            self._publish_market_event(
                "NEWS_IMPACT",
                f"新闻影响: {news_impact}",
                {'impact': news_impact}
            )

    def on_bar(self, bar: Bar):
        """处理K线数据"""
        # 更新市场状态
        self._update_market_state(bar)

        # 生成实时信号
        signal = self._analyze_bar_pattern(bar)
        if signal:
            self._generate_signal(signal['type'], signal['confidence'])

    def _update_market_state(self, bar: Bar):
        """更新市场状态"""
        # 计算波动率
        price_change = abs(bar.close - bar.open) / bar.open
        if price_change > Decimal('0.02'):
            self.market_state['volatility'] = 'HIGH'
        elif price_change > Decimal('0.01'):
            self.market_state['volatility'] = 'MEDIUM'
        else:
            self.market_state['volatility'] = 'LOW'

    def _analyze_bar_pattern(self, bar: Bar) -> dict:
        """分析K线模式"""
        # 简单的模式识别
        body_size = abs(bar.close - bar.open)
        upper_shadow = bar.high - max(bar.open, bar.close)
        lower_shadow = min(bar.open, bar.close) - bar.low

        # 十字星
        if body_size < (bar.high - bar.low) * Decimal('0.1'):
            return {'type': 'HOLD', 'confidence': 0.5}

        # 大阳线
        if bar.close > bar.open and body_size > (bar.high - bar.low) * Decimal('0.7'):
            return {'type': 'BUY', 'confidence': 0.6}

        # 大阴线
        if bar.close < bar.open and body_size > (bar.high - bar.low) * Decimal('0.7'):
            return {'type': 'SELL', 'confidence': 0.6}

        return None

    def _generate_signal(self, signal_type: str, confidence: float):
        """生成交易信号"""
        signal = {
            'type': signal_type,
            'confidence': confidence,
            'timestamp': self.clock.timestamp_ns(),
            'state': self.market_state.copy(),
        }

        self.signal_history.append(signal)

        # 发布信号事件
        event = SignalEvent(
            trader_id=self.trader_id,
            strategy_id=self.id,
            instrument_id="BTCUSDT.BINANCE",
            signal_type=signal_type,
            confidence=confidence,
            metadata=self.market_state.copy(),
        )

        self._handle_signal_event(event)

    def _calculate_total_exposure(self) -> Decimal:
        """计算总敞口"""
        # 简化计算
        return Decimal('50000')  # 模拟值

    def _calculate_drawdown(self) -> Decimal:
        """计算回撤"""
        # 简化计算
        return Decimal('0.05')  # 模拟5%回撤

    def _calculate_daily_pnl(self) -> Decimal:
        """计算日盈亏"""
        # 简化计算
        return Decimal('1000')  # 模拟1000 USDT

    def _calculate_win_rate(self) -> Decimal:
        """计算胜率"""
        # 简化计算
        return Decimal('0.6')  # 模拟60%

    def _calculate_sharpe_ratio(self) -> Decimal:
        """计算夏普比率"""
        # 简化计算
        return Decimal('1.5')  # 模拟1.5

    def _publish_market_event(self, event_type: str, message: str, data: dict):
        """发布市场事件"""
        event = CustomEvent(
            trader_id=self.trader_id,
            strategy_id=self.id,
            event_type=event_type,
            message=message,
            data=data,
        )
        self.event_timeline.append(event)

    def _publish_risk_event(self, risk_type: str, level: str, message: str):
        """发布风险事件"""
        event = RiskEvent(
            trader_id=self.trader_id,
            risk_type=risk_type,
            risk_level=level,
            message=message,
        )
        self.event_timeline.append(event)

    def _publish_performance_event(self, metric_type: str, value: float, period: str):
        """发布性能事件"""
        event = PerformanceEvent(
            trader_id=self.trader_id,
            strategy_id=self.id,
            metric_type=metric_type,
            metric_value=value,
            period=period,
        )
        self.event_timeline.append(event)

    def _publish_system_event(self, system_type: str, severity: str, message: str):
        """发布系统事件"""
        event = SystemEvent(
            trader_id=self.trader_id,
            system_type=system_type,
            severity=severity,
            message=message,
        )
        self.event_timeline.append(event)

    def on_stop(self):
        """策略停止"""
        self.log.info("综合策略停止")

        # 取消所有定时器
        for timer_name, timer_id in self.timers.items():
            if timer_id:
                self.clock.cancel_timer(timer_id)
                self.log.info(f"取消定时器: {timer_name}")

        # 生成最终报告
        self._generate_final_report()

    def _generate_final_report(self):
        """生成最终报告"""
        report = f"""
策略执行报告
================

运行时间: {self.clock.timestamp_ns() / 1_000_000_000:.0f} 秒
事件总数: {len(self.event_timeline)}
信号总数: {len(self.signal_history)}

市场状态
--------
趋势: {self.market_state['trend']}
波动率: {self.market_state['volatility']}
流动性: {self.market_state['liquidity']}

性能指标
--------
日盈亏: {self.performance_metrics['daily_pnl']} USDT
最大回撤: {self.performance_metrics['max_drawdown']:.1%}
胜率: {self.performance_metrics['win_rate']:.1%}
夏普比率: {self.performance_metrics['sharpe_ratio']}

最近事件
--------
"""

        # 添加最近5个事件
        for event in self.event_timeline[-5:]:
            report += f"- {event}\n"

        self.log.info(report)

6.5 调试和监控

6.5.1 事件追踪工具

"""
事件追踪和调试工具
"""

import json
from datetime import datetime
from typing import Dict, List, Optional
from collections import defaultdict, deque


class EventTracker:
    """事件追踪器"""

    def __init__(self, max_events: int = 10000):
        """
        初始化追踪器

        Parameters
        ----------
        max_events : int
            最大事件数
        """
        self.max_events = max_events
        self.events = deque(maxlen=max_events)
        self.event_stats = defaultdict(int)
        self.event_timeline = []

    def track_event(self, event: Event, source: str = "unknown"):
        """
        追踪事件

        Parameters
        ----------
        event : Event
            要追踪的事件
        source : str
            事件来源
        """
        event_info = {
            'id': str(event.id) if hasattr(event, 'id') else str(id(event)),
            'type': type(event).__name__,
            'timestamp': event.ts_event,
            'source': source,
            'data': self._serialize_event(event),
        }

        self.events.append(event_info)
        self.event_stats[event_info['type']] += 1

        # 添加到时间线
        self.event_timeline.append({
            'timestamp': event.ts_event,
            'event_type': event_info['type'],
            'source': source,
        })

    def _serialize_event(self, event: Event) -> Dict:
        """序列化事件数据"""
        data = {}

        # 提取常用属性
        if hasattr(event, 'trader_id'):
            data['trader_id'] = str(event.trader_id)
        if hasattr(event, 'strategy_id'):
            data['strategy_id'] = str(event.strategy_id)
        if hasattr(event, 'instrument_id'):
            data['instrument_id'] = str(event.instrument_id)

        # 对于自定义事件，添加额外属性
        if hasattr(event, 'metadata'):
            data['metadata'] = event.metadata

        return data

    def get_events_by_type(self, event_type: str) -> List[Dict]:
        """获取指定类型的事件"""
        return [e for e in self.events if e['type'] == event_type]

    def get_events_by_timerange(self, start_ns: int, end_ns: int) -> List[Dict]:
        """获取指定时间范围内的事件"""
        return [
            e for e in self.events
            if start_ns <= e['timestamp'] <= end_ns
        ]

    def get_event_sequence(self, event_id: str) -> List[Dict]:
        """获取事件序列"""
        sequence = []
        found = False

        for event in self.events:
            if event['id'] == event_id:
                found = True
            if found:
                sequence.append(event)

        return sequence

    def analyze_performance(self) -> Dict:
        """分析性能"""
        total_events = len(self.events)
        event_types = list(self.event_stats.keys())

        # 计算事件频率
        if len(self.event_timeline) > 1:
            duration = (self.event_timeline[-1]['timestamp'] -
                      self.event_timeline[0]['timestamp'])
            frequency = total_events / (duration / 1_000_000_000) if duration > 0 else 0
        else:
            frequency = 0

        return {
            'total_events': total_events,
            'unique_types': len(event_types),
            'event_frequency': frequency,
            'type_distribution': dict(self.event_stats),
            'memory_usage': len(self.events) * 100 / self.max_events,  # 百分比
        }

    def export_to_file(self, filename: str):
        """导出到文件"""
        data = {
            'export_time': datetime.now().isoformat(),
            'stats': self.analyze_performance(),
            'events': list(self.events),
            'timeline': self.event_timeline,
        }

        with open(filename, 'w') as f:
            json.dump(data, f, indent=2, default=str)

    def visualize_timeline(self):
        """可视化时间线（简化版）"""
        print("\n事件时间线:")
        print("-" * 80)

        last_time = None
        for event in self.event_timeline[-20:]:  # 显示最近20个事件
            time_str = datetime.fromtimestamp(
                event['timestamp'] / 1_000_000_000
            ).strftime('%H:%M:%S')

            if last_time:
                time_diff = (event['timestamp'] - last_time) / 1_000_000_000
                time_str += f" (+{time_diff:.1f}s)"

            print(f"{time_str}: {event['event_type']} from {event['source']}")
            last_time = event['timestamp']


class PerformanceProfiler:
    """性能分析器"""

    def __init__(self):
        """初始化分析器"""
        self.function_times = defaultdict(list)
        self.event_processing_times = defaultdict(list)

    def profile_function(self, func_name: str, execution_time_ns: int):
        """记录函数执行时间"""
        self.function_times[func_name].append(execution_time_ns)

    def profile_event_processing(
        self,
        event_type: str,
        processing_time_ns: int
    ):
        """记录事件处理时间"""
        self.event_processing_times[event_type].append(processing_time_ns)

    def get_function_stats(self) -> Dict:
        """获取函数统计"""
        stats = {}

        for func, times in self.function_times.items():
            if times:
                stats[func] = {
                    'count': len(times),
                    'total_time_ns': sum(times),
                    'avg_time_ns': sum(times) / len(times),
                    'max_time_ns': max(times),
                    'min_time_ns': min(times),
                }

        return stats

    def get_event_processing_stats(self) -> Dict:
        """获取事件处理统计"""
        stats = {}

        for event_type, times in self.event_processing_times.items():
            if times:
                stats[event_type] = {
                    'count': len(times),
                    'total_time_ns': sum(times),
                    'avg_time_ns': sum(times) / len(times),
                    'max_time_ns': max(times),
                }

        return stats

    def find_bottlenecks(self) -> List[Dict]:
        """找出性能瓶颈"""
        bottlenecks = []

        # 函数瓶颈
        func_stats = self.get_function_stats()
        for func, stats in func_stats.items():
            if stats['avg_time_ns'] > 1_000_000:  # 超过1ms
                bottlenecks.append({
                    'type': 'function',
                    'name': func,
                    'avg_time_ms': stats['avg_time_ns'] / 1_000_000,
                    'issue': '函数执行时间过长'
                })

        # 事件处理瓶颈
        event_stats = self.get_event_processing_stats()
        for event_type, stats in event_stats.items():
            if stats['avg_time_ns'] > 1_000_000:  # 超过1ms
                bottlenecks.append({
                    'type': 'event',
                    'name': event_type,
                    'avg_time_ms': stats['avg_time_ns'] / 1_000_000,
                    'issue': '事件处理时间过长'
                })

        # 按时间排序
        bottlenecks.sort(key=lambda x: x['avg_time_ms'], reverse=True)

        return bottlenecks


# 使用示例
def main():
    """主函数示例"""

    # 创建事件追踪器
    tracker = EventTracker()
    profiler = PerformanceProfiler()

    # 创建策略
    strategy = ComprehensiveStrategy()

    # 模拟运行
    import time
    start_time = time.time()

    # 模拟事件处理
    for i in range(100):
        # 创建模拟事件
        event = SignalEvent(
            trader_id=TraderId("TEST-001"),
            strategy_id=StrategyId("TEST-STRATEGY"),
            instrument_id="BTCUSDT.BINANCE",
            signal_type="BUY",
            confidence=0.8,
        )

        # 追踪事件
        start = time.time_ns()
        strategy._handle_signal_event(event)
        end = time.time_ns()

        tracker.track_event(event, "simulator")
        profiler.profile_event_processing("SignalEvent", end - start)

        # 模拟函数调用
        profiler.profile_function("market_analysis", time.time_ns() // 100)

    # 生成报告
    print("\n事件追踪报告:")
    print("=" * 50)

    stats = tracker.analyze_performance()
    print(f"总事件数: {stats['total_events']}")
    print(f"事件频率: {stats['event_frequency']:.2f} events/s")

    print("\n性能分析:")
    print("=" * 50)

    bottlenecks = profiler.find_bottlenecks()
    if bottlenecks:
        print("发现性能瓶颈:")
        for bottleneck in bottlenecks[:5]:
            print(f"- {bottleneck['name']}: {bottleneck['avg_time_ms']:.2f}ms "
                  f"({bottleneck['issue']})")
    else:
        print("未发现明显性能瓶颈")

    # 导出数据
    tracker.export_to_file("event_trace.json")

    # 显示时间线
    tracker.visualize_timeline()


if __name__ == "__main__":
    main()

6.6 下一步

在本章中，我们深入学习了：

NautilusTrader 的时间管理机制
定时器的使用和高级模式
事件驱动架构和最佳实践
事件处理和性能优化
调试和监控工具

在下一章中，我们将学习：

技术指标系统
内置指标的使用
自定义指标开发
指标级联和组合
指标性能优化

6.7 总结

关键要点

NautilusTrader 提供了灵活的时间管理系统
定时器支持复杂调度逻辑
事件驱动架构实现了松耦合设计
事件总线支持高性能的异步处理

最佳实践

合理使用定时器，避免过度调度
事件处理要快速，避免阻塞
使用事件追踪工具调试问题
定期分析性能瓶颈

6.8 参考资料

NautilusTrader第5章：基础回测引擎使用

Henry Lin — Wed, 01 Jul 2026 07:38:11 +0000

第5章：基础回测引擎使用

学习目标

通过本章学习，您将：

掌握 BacktestEngine 的核心配置
理解交易所和账户配置
学会添加交易工具和订单类型
运行完整的回测流程
分析和理解回测结果

5.1 BacktestEngine 概述

BacktestEngine 是 NautilusTrader 的核心回测组件，提供了高度可配置的回测环境。

5.1.1 架构图

┌─────────────────────────────────────────┐
│               BacktestEngine            │
├─────────────────────────────────────────┤
│  - 配置管理                              │
│  - 时间管理                              │
│  - 事件调度                              │
│  - 状态管理                              │
└─────────────────────────────────────────┘
            │
            ▼
┌─────────────────────────────────────────┐
│              核心组件                     │
├─────────────────────────────────────────┤
│  • ExecutionEngine (执行引擎)             │
│  • DataEngine (数据引擎)                 │
│  • RiskEngine (风险引擎)                 │
│  • Portfolio (投资组合)                  │
│  • Cache (缓存)                          │
│  • MessageBus (消息总线)                 │
└─────────────────────────────────────────┘

5.1.2 主要功能

模拟交易所：模拟真实交易所的撮合机制
订单管理：支持多种订单类型和执行算法
费用计算：精确计算手续费和滑点
风险控制：实时风险检查和限制
时间管理：支持不同时间模式和加速回测

5.2 基础配置

5.2.1 BacktestEngineConfig

"""
BacktestEngine 配置示例
"""

from nautilus_trader.config import BacktestEngineConfig, LoggingConfig
from nautilus_trader.model.identifiers import TraderId


def create_engine_config():
    """创建引擎配置"""
    config = BacktestEngineConfig(
        # 交易者标识
        trader_id=TraderId("BACKTESTER-001"),

        # 日志配置
        logging=LoggingConfig(
            log_level="INFO",        # 日志级别
            log_colors=True,         # 彩色日志
            print_to_stdout=True,    # 输出到控制台
            filter_messages=False,   # 不过滤消息
        ),

        # 时间配置
        time_increment_ns=1_000_000_000,  # 1秒时间增量（纳秒）
        bar_build_interval_ms=100,         # K线构建间隔（毫秒）

        # 其他选项
        debug=False,                        # 调试模式
        debug_messages=False,               # 调试消息
        benchmark_fees=False,               # 基准费用

        # 冲突解决模式
        fill_at_knowledgeable=False,        # 在不知情时成交
        reject_terminal_orders=False,      # 拒绝终端订单
        use_position_ids=False,             # 使用仓位ID
    )

    return config

5.2.2 创建和配置引擎

"""
完整的引擎配置示例
"""

from decimal import Decimal
from nautilus_trader.backtest.engine import BacktestEngine
from nautilus_trader.model.enums import OmsType, AccountType
from nautilus_trader.model.identifiers import Venue
from nautilus_trader.model.objects import Money
from nautilus_trader.model.currencies import USD, BTC
from nautilus_trader.test_kit.providers import TestInstrumentProvider


def create_and_configure_engine():
    """创建并配置回测引擎"""

    # 1. 创建配置
    config = BacktestEngineConfig(
        trader_id=TraderId("BACKTESTER-001"),
        logging=LoggingConfig(log_level="INFO"),
    )

    # 2. 创建引擎
    engine = BacktestEngine(config=config)

    # 3. 添加交易所（可以添加多个）
    add_venues(engine)

    # 4. 添加交易工具
    add_instruments(engine)

    return engine


def add_venues(engine: BacktestEngine):
    """添加交易所配置"""

    # Binance 配置
    BINANCE = Venue("BINANCE")
    engine.add_venue(
        venue=BINANCE,
        # 订单管理系统类型
        oms_type=OmsType.HEDGING,  # 对冲模式（允许同时持有多空）
        # 账户类型
        account_type=AccountType.MARGIN,  # 保证金账户
        # 起始资金
        starting_balances=[
            Money(1_000_000, USD),  # 100万USDT
            Money(20, BTC),         # 20 BTC
        ],
        # 基础货币
        base_currency=USD,
        # 杠杆配置
        leverage_real=False,       # 不使用真实杠杆
        # 费用配置（可选）
        account_id=None,           # 自动生成
        # 其他配置
        max_order_submit_rate=None,    # 最大订单提交速率
        max_order_cancel_rate=None,    # 最大订单取消速率
    )

    # 添加第二个交易所
    BYBIT = Venue("BYBIT")
    engine.add_venue(
        venue=BYBIT,
        oms_type=OmsType.NETTING,  # 净仓模式（单一方向）
        account_type=AccountType.CASH,  # 现金账户
        starting_balances=[Money(500_000, USD)],
        base_currency=USD,
    )

    print(f"已添加交易所: {BINANCE}, {BYBIT}")


def add_instruments(engine: BacktestEngine):
    """添加交易工具"""

    # BTC/USDT 永续合约
    btc_usdt_perp = TestInstrumentProvider.btcusdt_perp_binance()
    engine.add_instrument(btc_usdt_perp)

    # ETH/USDT 现货
    eth_usdt = TestInstrumentProvider.ethusdt_binance()
    engine.add_instrument(eth_usdt)

    # BTC/USD 期货
    btc_usd = TestInstrumentProvider.btcusdt_cme()
    engine.add_instrument(btc_usd)

    print(f"已添加交易工具: {len(engine.instruments())} 个")


def main():
    """主函数示例"""
    engine = create_and_configure_engine()
    print(f"引擎配置完成，支持 {len(cache.venues())} 个交易所")
    return engine


if __name__ == "__main__":
    main()

5.3 数据管理

5.3.1 添加历史数据

"""
数据管理示例
"""

from datetime import datetime, timedelta
from decimal import Decimal
import pandas as pd
from pathlib import Path

from nautilus_trader.model.data import Bar, QuoteTick, TradeTick
from nautilus_trader.model.data import BarType, BarSpecification
from nautilus_trader.model.enums import BarAggregation, PriceType
from nautilus_trader.model.identifiers import InstrumentId
from nautilus_trader.persistence.wranglers import BarDataWrangler
from nautilus_trader.test_kit.providers import TestInstrumentProvider


def add_bar_data(engine: BacktestEngine):
    """添加K线数据"""

    # 获取交易工具
    btc_usdt = TestInstrumentProvider.btcusdt_binance()

    # 定义K线类型
    bar_type = BarType(
        instrument_id=btc_usdt.id,
        bar_spec=BarSpecification(
            step=1,                        # 1根K线
            aggregation=BarAggregation.MINUTE,  # 1分钟
            price_type=PriceType.LAST,     # 最后成交价
        ),
        aggregation_source="BINANCE",
    )

    # 加载历史数据（示例）
    bars = load_historical_bars(bar_type, days=30)

    # 添加到引擎
    engine.add_data(bars)

    print(f"添加了 {len(bars)} 根 {bar_type} K线")


def add_tick_data(engine: BacktestEngine):
    """添加Tick数据"""

    # 获取交易工具
    btc_usdt = TestInstrumentProvider.btcusdt_binance()

    # 加载报价数据
    quotes = load_quote_ticks(btc_usdt.id, days=1)
    engine.add_data(quotes)

    # 加载成交数据
    trades = load_trade_ticks(btc_usdt.id, days=1)
    engine.add_data(trades)

    print(f"添加了 {len(quotes)} 个报价，{len(trades)} 个成交")


def add_mixed_data(engine: BacktestEngine):
    """添加混合数据（K线 + Tick）"""

    # 添加K线数据作为基础数据
    add_bar_data(engine)

    # 在关键时间点添加Tick数据
    tick_events = generate_key_tick_events()
    engine.add_data(tick_events)


def load_historical_bars(bar_type: BarType, days: int = 30) -> list[Bar]:
    """加载历史K线数据"""

    # 示例：生成模拟数据
    # 实际应用中应该从数据库或文件加载

    bars = []
    start_time = datetime.utcnow() - timedelta(days=days)

    for i in range(days * 24 * 60):  # 每分钟
        timestamp = start_time + timedelta(minutes=i)

        # 生成模拟OHLCV
        base_price = Decimal("50000")
        random_walk = Decimal(str((i % 100 - 50) * 10))

        bar = Bar(
            bar_type=bar_type,
            open=base_price + random_walk,
            high=base_price + random_walk + Decimal("100"),
            low=base_price + random_walk - Decimal("100"),
            close=base_price + random_walk + Decimal("50"),
            volume=Decimal(str(100 + i % 50)),
            ts_event=int(timestamp.timestamp() * 1e9),  # 纳秒时间戳
            ts_init=int(timestamp.timestamp() * 1e9),
        )

        bars.append(bar)

    return bars


def load_quote_ticks(instrument_id: InstrumentId, days: int = 1) -> list[QuoteTick]:
    """加载报价Tick数据"""
    quotes = []

    # 模拟报价数据
    for i in range(1000):
        price = Decimal("50000") + Decimal(str(i % 100))

        quote = QuoteTick(
            instrument_id=instrument_id,
            bid_price=price - Decimal("1"),
            ask_price=price + Decimal("1"),
            bid_size=Decimal("10"),
            ask_size=Decimal("10"),
            ts_event=1640995200000000000 + i * 1000000000,  # 每秒一个tick
            ts_init=1640995200000000000 + i * 1000000000,
        )

        quotes.append(quote)

    return quotes


def load_trade_ticks(instrument_id: InstrumentId, days: int = 1) -> list[TradeTick]:
    """加载成交Tick数据"""
    trades = []

    # 模拟成交数据
    for i in range(500):
        price = Decimal("50000") + Decimal(str(i % 100))

        trade = TradeTick(
            instrument_id=instrument_id,
            price=price,
            size=Decimal("1"),
            aggressor_side=AggressorSide.BUYER if i % 2 == 0 else AggressorSide.SELLER,
            trade_id=str(i),
            ts_event=1640995200000000000 + i * 2000000000,  # 每2秒一个trade
            ts_init=1640995200000000000 + i * 2000000000,
        )

        trades.append(trade)

    return trades

5.3.2 数据序列化

"""
数据序列化和持久化示例
"""

import pickle
import json
from pathlib import Path
from typing import Dict, Any

from nautilus_trader.serialization.serializert import Serializer


class DataSerializer:
    """数据序列化工具"""

    def __init__(self):
        """初始化序列化器"""
        self.serializer = Serializer()

    def serialize_to_pickle(self, data: list, file_path: Path):
        """序列化到Pickle文件"""
        with open(file_path, 'wb') as f:
            pickle.dump(data, f)

    def deserialize_from_pickle(self, file_path: Path) -> list:
        """从Pickle文件反序列化"""
        with open(file_path, 'rb') as f:
            return pickle.load(f)

    def serialize_to_json(self, bars: list[Bar], file_path: Path):
        """序列化K线到JSON文件"""
        data = []
        for bar in bars:
            data.append({
                'timestamp': bar.ts_event,
                'open': str(bar.open),
                'high': str(bar.high),
                'low': str(bar.low),
                'close': str(bar.close),
                'volume': str(bar.volume),
            })

        with open(file_path, 'w') as f:
            json.dump(data, f, indent=2)

    def export_to_dataframe(self, bars: list[Bar]) -> pd.DataFrame:
        """导出到DataFrame"""
        data = []
        for bar in bars:
            data.append({
                'timestamp': pd.to_datetime(bar.ts_event, unit='ns'),
                'open': float(bar.open),
                'high': float(bar.high),
                'low': float(bar.low),
                'close': float(bar.close),
                'volume': float(bar.volume),
            })

        return pd.DataFrame(data)


def main():
    """主函数示例"""
    # 创建序列化器
    serializer = DataSerializer()

    # 加载数据
    bars = load_historical_bars(bar_type, days=30)

    # 序列化到文件
    serializer.serialize_to_pickle(
        bars,
        Path("bars_2024_01.pkl")
    )

    serializer.serialize_to_json(
        bars,
        Path("bars_2024_01.json")
    )

    # 导出到DataFrame
    df = serializer.export_to_dataframe(bars)
    df.to_csv("bars_2024_01.csv", index=False)

    print("数据序列化完成")

5.4 策略添加和管理

5.4.1 添加策略

"""
策略管理示例
"""

from typing import List
from nautilus_trader.trading.strategy import Strategy
from nautilus_trader.model.enums import OrderSide
from nautilus_trader.model.orders import MarketOrder
from nautilus_trader.model.data import Bar
from nautilus_trader.model.data import BarType


class SimpleBuyStrategy(Strategy):
    """简单买入策略"""

    def __init__(self, bar_type: BarType, buy_threshold: float = 0.01):
        """
        初始化策略

        Parameters
        ----------
        bar_type : BarType
            K线类型
        buy_threshold : float
            买入阈值（1%）
        """
        super().__init__()
        self.bar_type = bar_type
        self.buy_threshold = buy_threshold
        self.last_price = None
        self.bought = False

    def on_start(self):
        """策略启动"""
        self.log.info(f"策略启动: {self.bar_type}")
        self.subscribe_bars(self.bar_type)

    def on_bar(self, bar: Bar):
        """处理K线"""
        current_price = float(bar.close)

        if self.last_price is None:
            self.last_price = current_price
            return

        # 计算涨跌幅
        change = (current_price - self.last_price) / self.last_price

        # 如果涨幅超过阈值且未买入，则买入
        if change > self.buy_threshold and not self.bought:
            self._buy(bar)
        elif change < -self.buy_threshold and self.bought:
            self._sell(bar)

        self.last_price = current_price

    def _buy(self, bar: Bar):
        """执行买入"""
        order = MarketOrder(
            trader_id=self.trader_id,
            strategy_id=self.id,
            instrument_id=self.bar_type.instrument_id,
            order_side=OrderSide.BUY,
            quantity=Decimal("0.1"),
            tags="SIMPLE_BUY",
        )

        self.submit_order(order)
        self.bought = True
        self.log.info(f"买入: {bar.close}")

    def _sell(self, bar: Bar):
        """执行卖出"""
        order = MarketOrder(
            trader_id=self.trader_id,
            strategy_id=self.id,
            instrument_id=self.bar_type.instrument_id,
            order_side=OrderSide.SELL,
            quantity=Decimal("0.1"),
            tags="SIMPLE_SELL",
        )

        self.submit_order(order)
        self.bought = False
        self.log.info(f"卖出: {bar.close}")


def add_strategies(engine: BacktestEngine):
    """添加策略到引擎"""

    # 策略1：BTC/USDT 1分钟
    strategy1 = SimpleBuyStrategy(
        bar_type=BarType(
            instrument_id=InstrumentId.from_str("BTCUSDT.BINANCE"),
            bar_spec=BarSpecification(
                step=1,
                aggregation=BarAggregation.MINUTE,
                price_type=PriceType.LAST,
            ),
            aggregation_source="BINANCE",
        ),
        buy_threshold=0.005,  # 0.5%
    )
    engine.add_strategy(strategy1)

    # 策略2：ETH/USDT 5分钟
    strategy2 = SimpleBuyStrategy(
        bar_type=BarType(
            instrument_id=InstrumentId.from_str("ETHUSDT.BINANCE"),
            bar_spec=BarSpecification(
                step=5,
                aggregation=BarAggregation.MINUTE,
                price_type=PriceType.LAST,
            ),
            aggregation_source="BINANCE",
        ),
        buy_threshold=0.003,  # 0.3%
    )
    engine.add_strategy(strategy2)

    print(f"添加了 2 个策略")


class MultiAssetStrategy(Strategy):
    """多资产策略"""

    def __init__(self, bar_types: List[BarType]):
        """
        初始化多资产策略

        Parameters
        ----------
        bar_types : List[BarType]
            多个K线类型
        """
        super().__init__()
        self.bar_types = bar_types
        self.prices = {}

    def on_start(self):
        """策略启动"""
        self.log.info("多资产策略启动")

        for bar_type in self.bar_types:
            self.subscribe_bars(bar_type)

    def on_bar(self, bar: Bar):
        """处理K线"""
        instrument_id = str(bar.bar_type.instrument_id)
        self.prices[instrument_id] = float(bar.close)

        # 如果有至少两个资产的价格
        if len(self.prices) >= 2:
            self._check_arbitrage()

    def _check_arbitrage(self):
        """检查套利机会"""
        # 这里可以实现简单的套利逻辑
        pass


def add_multi_asset_strategy(engine: BacktestEngine):
    """添加多资产策略"""

    bar_types = [
        BarType(
            instrument_id=InstrumentId.from_str("BTCUSDT.BINANCE"),
            bar_spec=BarSpecification(
                step=1,
                aggregation=BarAggregation.MINUTE,
                price_type=PriceType.LAST,
            ),
            aggregation_source="BINANCE",
        ),
        BarType(
            instrument_id=InstrumentId.from_str("ETHUSDT.BINANCE"),
            bar_spec=BarSpecification(
                step=1,
                aggregation=BarAggregation.MINUTE,
                price_type=PriceType.LAST,
            ),
            aggregation_source="BINANCE",
        ),
    ]

    strategy = MultiAssetStrategy(bar_types)
    engine.add_strategy(strategy)

    print(f"添加了多资产策略，监控 {len(bar_types)} 个资产")

5.5 运行回测

5.5.1 基本回测流程

"""
完整的回测执行示例
"""

from datetime import datetime, timedelta
from decimal import Decimal
import time


def run_simple_backtest():
    """运行简单回测"""

    print("开始回测...")
    start_time = time.time()

    # 1. 创建引擎
    config = BacktestEngineConfig(
        trader_id=TraderId("BACKTESTER-001"),
        logging=LoggingConfig(log_level="INFO"),
    )
    engine = BacktestEngine(config=config)

    # 2. 添加交易所
    engine.add_venue(
        venue=Venue("BINANCE"),
        oms_type=OmsType.HEDGING,
        account_type=AccountType.MARGIN,
        starting_balances=[Money(100_000, USDT)],
        base_currency=USDT,
    )

    # 3. 添加交易工具
    btc_usdt = TestInstrumentProvider.btcusdt_binance()
    engine.add_instrument(btc_usdt)

    # 4. 添加数据
    bar_type = BarType(
        instrument_id=btc_usdt.id,
        bar_spec=BarSpecification(
            step=1,
            aggregation=BarAggregation.MINUTE,
            price_type=PriceType.LAST,
        ),
        aggregation_source="BINANCE",
    )
    bars = load_historical_bars(bar_type, days=30)
    engine.add_data(bars)

    # 5. 添加策略
    strategy = SimpleBuyStrategy(bar_type)
    engine.add_strategy(strategy)

    # 6. 运行回测
    print("执行回测...")
    engine.run()

    # 7. 获取结果
    end_time = time.time()
    duration = end_time - start_time

    print(f"\n回测完成！")
    print(f"执行时间: {duration:.2f} 秒")
    print(f"处理K线: {len(bars)} 根")

    # 8. 分析结果
    analyze_results(engine)

    # 9. 清理资源
    engine.dispose()

    return engine


def run_parameter_sweep():
    """运行参数扫描"""

    # 参数列表
    buy_thresholds = [0.003, 0.005, 0.01, 0.02]

    results = []

    for threshold in buy_thresholds:
        print(f"\n测试参数: buy_threshold = {threshold}")

        # 创建引擎
        config = BacktestEngineConfig(
            trader_id=TraderId(f"BACKTESTER-{threshold}"),
            logging=LoggingConfig(log_level="ERROR"),  # 减少日志
        )
        engine = BacktestEngine(config=config)

        # 配置引擎（简化）
        configure_engine(engine)

        # 创建策略
        strategy = SimpleBuyStrategy(bar_type, buy_threshold=threshold)
        engine.add_strategy(strategy)

        # 运行回测
        engine.run()

        # 收集结果
        result = analyze_results_brief(engine)
        result['threshold'] = threshold
        results.append(result)

        # 清理
        engine.dispose()

    # 打印对比结果
    print("\n参数扫描结果:")
    print("-" * 80)
    for result in results:
        print(f"阈值: {result['threshold']:.3f} | "
              f"交易次数: {result['trade_count']:3d} | "
              f"胜率: {result['win_rate']:5.1%} | "
              f"总盈亏: {result['total_pnl']:8.2f}")

    return results


def run_multi_strategy_backtest():
    """运行多策略回测"""

    print("开始多策略回测...")

    # 创建引擎
    config = BacktestEngineConfig(
        trader_id=TraderId("MULTI-STRATEGY-001"),
        logging=LoggingConfig(log_level="INFO"),
    )
    engine = BacktestEngine(config=config)

    # 添加多个交易所
    engine.add_venue(
        venue=Venue("BINANCE"),
        oms_type=OmsType.HEDGING,
        account_type=AccountType.MARGIN,
        starting_balances=[Money(100_000, USDT)],
        base_currency=USDT,
    )

    engine.add_venue(
        venue=Venue("BYBIT"),
        oms_type=OmsType.NETTING,
        account_type=AccountType.CASH,
        starting_balances=[Money(50_000, USDT)],
        base_currency=USDT,
    )

    # 添加多个交易工具
    instruments = [
        TestInstrumentProvider.btcusdt_binance(),
        TestInstrumentProvider.ethusdt_binance(),
    ]

    for instrument in instruments:
        engine.add_instrument(instrument)

    # 添加数据
    for instrument in instruments:
        bar_type = BarType(
            instrument_id=instrument.id,
            bar_spec=BarSpecification(
                step=1,
                aggregation=BarAggregation.MINUTE,
                price_type=PriceType.LAST,
            ),
            aggregation_source=instrument.id.venue.value,
        )
        bars = load_historical_bars(bar_type, days=30)
        engine.add_data(bars)

    # 添加多个策略
    strategies = [
        SimpleBuyStrategy(bar_type_1, buy_threshold=0.005),
        SimpleBuyStrategy(bar_type_2, buy_threshold=0.003),
        MultiAssetStrategy([bar_type_1, bar_type_2]),
    ]

    for strategy in strategies:
        engine.add_strategy(strategy)

    # 运行回测
    start_time = time.time()
    engine.run()
    duration = time.time() - start_time

    print(f"\n多策略回测完成！")
    print(f"执行时间: {duration:.2f} 秒")
    print(f"策略数量: {len(strategies)}")

    # 分析结果
    analyze_multi_strategy_results(engine)

    # 清理
    engine.dispose()

    return engine

5.6 结果分析

5.6.1 获取和分析结果

"""
回测结果分析
"""

from typing import Dict, Any
import pandas as pd


def analyze_results(engine: BacktestEngine):
    """分析回测结果"""

    # 获取结果对象
    result = engine.get_result()

    # 基本统计
    print("\n=== 基本统计 ===")
    print(f"运行时间: {result.run_duration_seconds:.2f} 秒")
    print(f"处理事件: {result.total_events}")
    print(f"执行命令: {result.total_commands}")

    # 交易统计
    print("\n=== 交易统计 ===")
    stats = result.stats_pnls()

    for key, value in stats.items():
        if isinstance(value, Decimal):
            print(f"{key}: {float(value):.2f}")
        else:
            print(f"{key}: {value}")

    # 仓位信息
    print("\n=== 仓位信息 ===")
    print(f"当前仓位: {len(result.positions_open)} 个")
    print(f"已平仓位: {len(result.positions_closed)} 个")

    # 已平仓位详情
    if result.positions_closed:
        print("\n已平仓位详情:")
        for position in result.positions_closed[:5]:  # 只显示前5个
            print(f"  {position}")
            print(f"    开仓: {position.opened_timestamp}")
            print(f"    平仓: {position.closed_timestamp}")
            print(f"    盈亏: {position.realized_pnl}")
            print(f"    手续费: {position.total_commission}")

    # 订单统计
    print("\n=== 订单统计 ===")
    print(f"总订单数: {len(result.orders)}")

    # 按状态统计订单
    order_status_counts = {}
    for order in result.orders:
        status = order.status
        order_status_counts[status] = order_status_counts.get(status, 0) + 1

    for status, count in order_status_counts.items():
        print(f"  {status}: {count}")


def analyze_results_brief(engine: BacktestEngine) -> Dict[str, Any]:
    """简要分析结果"""

    result = engine.get_result()
    stats = result.stats_pnls()

    # 计算交易统计
    trade_count = len(result.positions_closed)
    win_count = sum(1 for pos in result.positions_closed
                    if pos.realized_pnl > 0)

    return {
        'trade_count': trade_count,
        'win_rate': win_count / trade_count if trade_count > 0 else 0,
        'total_pnl': float(stats.get('Total PnL', 0)),
        'realized_pnl': float(stats.get('Realized PnL', 0)),
        'max_drawdown': float(stats.get('Max Drawdown', 0)),
        'sharpe_ratio': float(stats.get('Sharpe Ratio', 0)),
    }


def analyze_multi_strategy_results(engine: BacktestEngine):
    """分析多策略结果"""

    result = engine.get_result()

    # 获取所有策略
    strategies = engine.trader.strategies()

    print("\n=== 策略表现 ===")
    for strategy_id, strategy in strategies.items():
        # 这里需要更详细的策略级别分析
        # 由于框架限制，我们使用整体统计
        pass

    # 按交易所统计
    print("\n=== 交易所统计 ===")
    venue_stats = {}

    for position in result.positions_closed:
        venue = position.instrument_id.venue.value
        if venue not in venue_stats:
            venue_stats[venue] = {
                'count': 0,
                'total_pnl': 0,
                'trades': []
            }

        venue_stats[venue]['count'] += 1
        venue_stats[venue]['total_pnl'] += float(position.realized_pnl)
        venue_stats[venue]['trades'].append(float(position.realized_pnl))

    for venue, stats in venue_stats.items():
        win_rate = sum(1 for pnl in stats['trades'] if pnl > 0) / len(stats['trades'])
        print(f"{venue}:")
        print(f"  交易次数: {stats['count']}")
        print(f"  总盈亏: {stats['total_pnl']:.2f}")
        print(f"  胜率: {win_rate:.1%}")


def export_results_to_csv(engine: BacktestEngine, filename: str):
    """导出结果到CSV"""

    result = engine.get_result()

    # 导出交易记录
    trades_data = []

    for position in result.positions_closed:
        trades_data.append({
            'instrument': str(position.instrument_id),
            'opened': position.opened_timestamp,
            'closed': position.closed_timestamp,
            'duration': str(position.open_duration),
            'side': 'LONG' if position.is_long else 'SHORT',
            'entry_price': float(position.avg_px_open),
            'exit_price': float(position.avg_px_close),
            'quantity': float(position.quantity),
            'pnl': float(position.realized_pnl),
            'commissions': float(position.total_commission),
            'return_pct': (float(position.realized_pnl) /
                         float(position.quantities_executed) * 100),
        })

    df = pd.DataFrame(trades_data)
    df.to_csv(f"{filename}_trades.csv", index=False)

    # 导出统计摘要
    stats = result.stats_pnls()
    stats_df = pd.DataFrame([stats])
    stats_df.to_csv(f"{filename}_stats.csv", index=False)

    print(f"结果已导出到 {filename}_*.csv")


def create_performance_report(engine: BacktestEngine) -> str:
    """创建性能报告"""

    result = engine.get_result()
    stats = analyze_results_brief(engine)

    report = f"""
回测性能报告
================

基本信息
--------
运行时间: {result.run_duration_seconds:.2f} 秒
处理事件: {result.total_events:,}

交易表现
--------
总交易次数: {stats['trade_count']}
胜率: {stats['win_rate']:.1%}
总盈亏: {stats['total_pnl']:,.2f} USDT
已实现盈亏: {stats['realized_pnl']:,.2f} USDT
最大回撤: {stats['max_drawdown']:,.2f} USDT
夏普比率: {stats['sharpe_ratio']:.2f}

风险指标
--------
盈利交易: {sum(1 for pos in result.positions_closed if pos.realized_pnl > 0)}
亏损交易: {sum(1 for pos in result.positions_closed if pos.realized_pnl < 0)}
平均盈亏: {stats['realized_pnl'] / stats['trade_count'] if stats['trade_count'] > 0 else 0:.2f} USDT

仓位信息
--------
当前持仓: {len(result.positions_open)} 个
已平仓位: {len(result.positions_closed)} 个
"""

    return report


def main():
    """主函数示例"""

    # 运行单个回测
    engine = run_simple_backtest()

    # 创建报告
    report = create_performance_report(engine)
    print(report)

    # 导出结果
    export_results_to_csv(engine, "backtest_2024_01")

    # 运行参数扫描
    print("\n" + "=" * 50)
    print("运行参数扫描...")
    results = run_parameter_sweep()

    # 找出最佳参数
    best_result = max(results, key=lambda x: x['total_pnl'])
    print(f"\n最佳参数: buy_threshold = {best_result['threshold']}")
    print(f"最高盈亏: {best_result['total_pnl']:,.2f} USDT")


if __name__ == "__main__":
    main()

5.7 高级功能

5.7.1 自定义费用模型

"""
自定义费用模型
"""

from nautilus_trader.model.fees import FeeModel
from nautilus_trader.model.objects import Money
from nautilus_trader.model.objects import Price
from nautilus_trader.model.objects import Quantity
from decimal import Decimal


class CustomFeeModel(FeeModel):
    """自定义费用模型"""

    def __init__(self, base_rate: Decimal = Decimal("0.001")):
        """
        初始化费用模型

        Parameters
        ----------
        base_rate : Decimal
            基础费率（0.1%）
        """
        self.base_rate = base_rate
        self.volume_discounts = {
            Decimal("1000"): Decimal("0.0008"),    # 1000+ USDT: 0.08%
            Decimal("10000"): Decimal("0.0006"),   # 10000+ USDT: 0.06%
            Decimal("100000"): Decimal("0.0004"),  # 100000+ USDT: 0.04%
        }

    def calculate(
        self,
        price: Price,
        quantity: Quantity,
        side: str,
        instrument_id: InstrumentId,
    ) -> Money:
        """
        计算费用

        Parameters
        ----------
        price : Price
            价格
        quantity : Quantity
            数量
        side : str
            方向
        instrument_id : InstrumentId
            交易工具ID

        Returns
        -------
        Money
            费用金额
        """
        # 计算交易额
        notional_value = price * quantity

        # 查找适用的费率
        rate = self.base_rate

        for threshold, discount_rate in self.volume_discounts.items():
            if notional_value >= threshold:
                rate = discount_rate
                break

        # 计算费用
        fee_amount = notional_value * rate

        # VIP折扣（示例）
        if is_vip_customer(instrument_id.venue):
            fee_amount *= Decimal("0.9")  # 10% VIP折扣

        return Money(fee_amount, instrument_id.quote_currency)


def configure_custom_fees(engine: BacktestEngine):
    """配置自定义费用"""

    # 为特定交易所配置费用模型
    venue = Venue("BINANCE")
    engine.add_fee_model(
        venue=venue,
        instrument_id=None,  # 应用于所有交易工具
        fee_model=CustomFeeModel(Decimal("0.001")),
    )

    print("已配置自定义费用模型")

5.7.2 滑点模型

"""
滑点模拟
"""

from nautilus_trader.model.slippage import SlippageModel
from random import uniform
from decimal import Decimal


class RandomSlippageModel(SlippageModel):
    """随机滑点模型"""

    def __init__(self, max_slippage: Decimal = Decimal("0.001")):
        """
        初始化滑点模型

        Parameters
        ----------
        max_slippage : Decimal
            最大滑点（0.1%）
        """
        self.max_slippage = max_slippage

    def calculate(
        self,
        order: Order,
        fill_price: Price,
        instrument_id: InstrumentId,
    ) -> Price:
        """
        计算滑点后的价格

        Parameters
        ----------
        order : Order
            订单
        fill_price : Price
            填充价格
        instrument_id : InstrumentId
            交易工具ID

        Returns
        -------
        Price
            滑点后的价格
        """
        # 生成随机滑点
        slippage_rate = Decimal(str(uniform(0, float(self.max_slippage))))

        # 买单价格上移，卖单价格下移
        if order.side == OrderSide.BUY:
            slippage_amount = fill_price * slippage_rate
            adjusted_price = fill_price + slippage_amount
        else:
            slippage_amount = fill_price * slippage_rate
            adjusted_price = fill_price - slippage_amount

        return Price(adjusted_price, fill_price.precision)


def configure_slippage(engine: BacktestEngine):
    """配置滑点模型"""

    # 为交易所添加滑点模型
    engine.add_slippage_model(
        venue=Venue("BINANCE"),
        slippage_model=RandomSlippageModel(Decimal("0.0005")),  # 0.05%
    )

    print("已配置滑点模型")

5.8 最佳实践

5.8.1 性能优化

"""
性能优化技巧
"""

import multiprocessing
from typing import List
from concurrent.futures import ProcessPoolExecutor


def optimized_backtest(param_set: List[Dict]) -> List[Dict]:
    """优化的批量回测"""

    # 使用进程池并行运行
    with ProcessPoolExecutor(max_workers=multiprocessing.cpu_count()) as executor:
        results = list(executor.map(run_single_backtest, param_set))

    return results


def run_single_backtest(params: Dict) -> Dict:
    """运行单个回测（用于并行）"""

    # 创建最小日志配置
    config = BacktestEngineConfig(
        trader_id=TraderId(f"BACKTEST-{params['id']}"),
        logging=LoggingConfig(log_level="ERROR"),
    )

    engine = BacktestEngine(config=config)

    # 最小配置
    engine.add_venue(
        venue=Venue("BINANCE"),
        oms_type=OmsType.NETTING,
        account_type=AccountType.CASH,
        starting_balances=[Money(params['balance'], USDT)],
        base_currency=USDT,
    )

    # 添加数据和策略
    # ...

    # 运行
    engine.run()

    # 收集结果
    result = analyze_results_brief(engine)
    result['params'] = params

    # 清理
    engine.dispose()

    return result


def memory_efficient_backtest():
    """内存高效的回测"""

    config = BacktestEngineConfig(
        trader_id=TraderId("EFFICIENT-001"),
        # 限制日志数量
        logging=LoggingConfig(
            log_level="WARNING",
            log_colors=False,
            print_to_stdout=False,
        ),
        # 启用内存优化
        debug=False,
        debug_messages=False,
    )

    engine = BacktestEngine(config=config)

    # 分批加载数据
    for day in range(30):  # 30天数据
        bars = load_bars_for_day(day)
        engine.add_data(bars)

        # 可选：定期清理内存
        if day % 7 == 0:
            engine.cache._flush()

    return engine

5.8.2 调试技巧

"""
回测调试工具
"""

import logging
from datetime import datetime


class BacktestDebugger:
    """回测调试器"""

    def __init__(self, engine: BacktestEngine):
        """初始化调试器"""
        self.engine = engine
        self.order_log = []
        self.fill_log = []

    def enable_verbose_logging(self):
        """启用详细日志"""

        # 创建文件处理器
        file_handler = logging.FileHandler("backtest_debug.log")
        file_handler.setLevel(logging.DEBUG)

        # 添加到引擎日志器
        engine.logger.addHandler(file_handler)
        engine.logger.setLevel(logging.DEBUG)

    def log_order_event(self, event):
        """记录订单事件"""
        self.order_log.append({
            'time': datetime.utcnow(),
            'type': type(event).__name__,
            'order_id': str(event.order_id),
            'data': str(event),
        })

    def analyze_order_flow(self):
        """分析订单流"""

        print("\n=== 订单流分析 ===")

        # 订单状态变化
        order_states = {}
        for log in self.order_log:
            order_id = log['order_id']
            if order_id not in order_states:
                order_states[order_id] = []
            order_states[order_id].append({
                'time': log['time'],
                'event': log['type'],
            })

        # 找出异常订单
        for order_id, events in order_states.items():
            if len(events) > 10:  # 太多状态变化
                print(f"异常订单: {order_id}")
                for event in events:
                    print(f"  {event['time']}: {event['event']}")

    def save_debug_data(self):
        """保存调试数据"""

        import json
        import pandas as pd

        # 保存订单日志
        df_orders = pd.DataFrame(self.order_log)
        df_orders.to_csv("debug_orders.csv", index=False)

        # 保存填充日志
        df_fills = pd.DataFrame(self.fill_log)
        df_fills.to_csv("debug_fills.csv", index=False)

        print("调试数据已保存")

5.9 下一步

在本章中，我们学习了：

BacktestEngine 的配置和使用
交易所和账户管理
数据添加和管理
策略集成
结果分析
性能优化技巧

在下一章中，我们将学习：

时间管理机制
事件驱动编程
定时器和调度
并发处理
高级事件模式

5.10 总结

关键要点

BacktestEngine 提供了完整的回测环境
支持多交易所、多资产、多策略回测
高度可配置，支持自定义费用和滑点
提供详细的结果分析工具

最佳实践

合理配置日志级别以提高性能
使用批量处理处理大量数据
定期清理内存
保存和分析调试信息

5.11 参考资料

美股-后端服务模块

Henry Lin — Sun, 14 Jun 2026 09:19:45 +0000

后端服务模块（backend/）

FastAPI 后端负责加载离线快照数据，提供查询、过滤、排序等 API，支持实时数据推送、预警系统、信号产品系统、因子回测和宏观/板块数据接口。

目录结构

backend/
├── main.py               # FastAPI 应用入口，路由注册，生命周期管理
├── prod_app.py           # 生产环境应用（含静态文件服务）
├── loader.py             # 数据加载、缓存、查询引擎
├── meta.py               # 元数据与筛选分组定义
├── presets.py            # 多时间周期预设策略定义
├── market_hours.py       # 美股交易时段与假日日历
│
│  # ── 实时数据 ──
├── rt_api.py             # 实时数据 API 路由
├── rt_hub.py             # 实时推送中心（WebSocket 编排）
├── rt_ws.py              # WebSocket 端点（设备注册、订阅管理）
├── rt_aliyun_ticker.py   # 阿里云行情后台 Ticker
│
│  # ── 预警系统 ──
├── alerts_api.py         # 预警 API 路由（CRUD + 事件查询）
├── alerts_db.py          # SQLite 存储层（规则、事件、设备）
├── alert_engine.py       # 预警引擎（规则评估、状态管理）
├── alert_dispatcher.py   # 预警分发器（WebSocket 推送）
├── alert_evaluators.py   # 规则评估器（6 种预警类型）
├── alert_models.py       # 数据结构（AlertRule, AlertEvent 等）
├── alerts_presets.py     # 预警模板（6 种预设规则）
│
│  # ── 信号产品系统 ──
├── signal_product_api.py    # 信号产品 API 路由（候选池、信号、绩效、模拟交易）
├── signal_product_models.py # 信号数据模型（Candidate/Signal/Outcome/Position）
├── signal_product_db.py     # 信号 SQLite 持久化（4 表：candidates/signals/outcomes/positions）
├── signal_entry_engine.py   # 盘中信号入场评估引擎
│
│  # ── 回测系统 ──
├── backtest_api.py       # 因子回测 API 路由（IC 评估、绩效分析）
│
│  # ── 数据接口 ──
├── screener_api.py       # 选股器 API（排名、评分）
├── macro.py              # 宏观经济指标接口
├── sectors.py            # 板块轮动数据接口
│
│  # ── 工具 ──
├── app/factors/
│   └── date_utils.py     # 日期格式工具（YYYYMMDD 标准化）
└── test_*.py             # 测试文件

启动流程

应用启动
  ↓
加载 config.yaml（实时数据 + 预警配置）
  ↓
load_snapshot()  → 缓存到 _SNAPSHOT_CACHE
load_history()   → 缓存到 _HISTORY_CACHE
  ↓
初始化 RTManager（实时数据管理器）
初始化 RTHub（推送中心）
初始化 AlertEngine + AlertDispatcher（预警引擎）
初始化 DeviceRegistry（WebSocket 设备注册）
初始化 SignalProductRepository（信号产品 SQLite）
  ↓
注册路由 + CORS 中间件
  ↓
可选：启动 Aliyun Ticker（后台定时获取行情）
  ↓
uvicorn 启动，监听 8000 端口

各文件详解

main.py — 应用入口

FastAPI 应用的主入口，管理生命周期和路由。

生命周期：

启动时：加载配置、初始化实时数据管理器、RTHub 推送中心、预警引擎、信号产品仓库、加载快照和历史数据
关闭时：停止 Aliyun Ticker、清理资源

API 端点：

端点	方法	说明	参数
`/api/meta`	GET	返回筛选元数据（分组、标签、统计）	—
`/api/stocks`	GET	查询股票列表	filters, conditions, sort_by, sort_asc, offset, limit
`/api/pattern-stats`	GET	形态统计（触发次数、胜率、收益率）	—
`/api/combo-stats`	GET	多信号组合统计	—
`/api/stocks/{symbol}/history`	GET	单只股票历史 K 线	range (3m/6m/1y/2y)
`/api/stock/{symbol}`	GET	单只股票快照数据	—
`/api/stocks/{symbol}/fundamentals`	GET	单只股票基本面数据	—
`/api/presets`	GET	多时间周期预设策略列表	—

loader.py — 数据引擎

核心数据模块，负责 parquet 加载、内存缓存和查询处理。

缓存机制：

data.parquet  →  _SNAPSHOT_CACHE (DataFrame)  →  内存常驻
history.parquet → _HISTORY_CACHE (DataFrame)  →  内存常驻

全局变量 _SNAPSHOT_CACHE / _HISTORY_CACHE 实现懒加载
首次访问时从磁盘加载，后续直接读内存

查询流程：

query_stocks(filters, conditions, sort_by, sort_asc, offset, limit)
  ↓
_apply_filters()      →  按标志列筛选（value=1）
  ↓
_parse_conditions()   →  解析 JSON 条件（支持 <, <=, >, >=, =）
  ↓
_apply_conditions()   →  应用连续值条件过滤
  ↓
排序 + 分页
  ↓
_to_records()         →  转换为 JSON 格式返回

列分类：

类别	说明	示例
REQUIRED_COLUMNS	必需字段	trade_date, symbol, name
CONTINUOUS_FILTER_COLUMNS	连续值列	ret_1d, ma5, forward_pe, piotroski_score, value_quality_score, altman_z_score, rsi_14, macd_dif, mfi_14...
SORTABLE_COLUMNS	可排序字段	signal_score, market_cap, piotroski_score, value_quality_score, altman_z_score...
PATTERN_FLAG_COLUMNS	形态标志列	96+ 个二值指标（含 ind_macd_bull_cross, ind_rsi_oversold, ind_bb_squeeze 等）

meta.py — 元数据定义

定义前端的筛选分组结构和统计信息。

筛选分组：

分组 ID	名称	指标数	示例指标
single_candle	单 K 形态	7	bull_candle, hammer, doji
double_candle	双 K 形态	3	bullish_engulfing, inside_bar
triple_candle	三 K 形态	2	three_white_soldiers
trend_structure	趋势结构	6	ma_bull, platform_breakout
volume_volatility	量价波动	4	volume_breakout, vol_contraction
momentum	动量	4	break_high_20, consec_up_3
valuation	估值指标	5	forward_pe, peg_ratio
financials	财务指标	8	roe, profit_margin, revenue_growth
fundamental_quality	基本面质量	3	fund_piotroski_strong, fund_value_quality, risk_financial_distress
advanced_indicators	高级技术指标	20+	ind_macd_bull_cross, ind_rsi_oversold, ind_bb_squeeze, ind_mfi_oversold 等

核心函数：

build_meta_response() — 构建完整的元数据响应（含统计）
build_pattern_stats_response() — 形态命中统计和标签
_count_indicator_hits() — 计算各指标命中数量

signal_product_api.py — 信号产品 API

信号产品系统的 REST API，覆盖候选池、入场信号、绩效追踪和模拟交易全流程。

端点	方法	说明
`/api/signals/status`	GET	信号系统状态（活跃信号数、今日候选数、日期）
`/api/signals/candidates`	GET	每日候选股列表（支持按日期/自选股/股票代码过滤）
`/api/signals/active`	GET	当前活跃（持仓中）信号列表
`/api/signals/history`	GET	信号历史（分页，含已平仓信号）
`/api/signals/performance/summary`	GET	聚合绩效指标（胜率、收益率、Sharpe）
`/api/signals/evaluate`	GET	用实时数据评估今日全部候选股
`/api/signals/evaluate/{symbol}`	GET	用实时数据评估单只候选股
`/api/signals/portfolio`	GET	模拟持仓列表
`/api/signals/portfolio/follow`	POST	跟随信号开仓（模拟交易）
`/api/signals/portfolio/{position_id}/close`	POST	平仓（模拟交易）

signal_product_models.py — 信号数据模型

信号产品管线的全部数据结构。

核心数据类：

数据类	说明	关键字段
`CandidateRecord`	每日候选股	candidate_id, symbol, rank, offline_score, 各维度评分, 风险标签
`SignalRecord`	盘中入场信号	signal_id, candidate_id, symbol, entry/stop/target 价格, trigger_reasons
`SignalOutcome`	信号绩效追踪	forward_return_1d/3d/5d/10d, MFE, MAE, hit_stop, hit_target
`PaperPosition`	模拟交易仓位	symbol, entry_price, quantity, stop/target, unrealized_pnl

CandidateRecord 风险门控：

HARD_RISK_FLAGS = {"low_liquidity", "abnormal_volatility", "weekly_downtrend"}
is_eligible() — 检查评分阈值和无硬性风险标记

signal_product_db.py — 信号 SQLite 存储

信号产品系统的持久化层，使用 SQLite WAL 模式。

数据表：

表	用途	关键索引
candidates	每日候选股打分结果	date, symbol
signals	盘中入场信号	date, symbol, status
signal_outcomes	后验绩效（MFE/MAE/前向收益）	signal_id, symbol
paper_positions	模拟交易仓位	symbol, status

特性：WAL 模式并发、JSON 序列化、UPSERT 更新。

signal_entry_engine.py — 盘中信号入场引擎

评估是否为候选股生成盘中入场信号。

4 重入场门控（全部通过才生成信号）：

门控	函数	条件
VWAP 门控	`_gate_price_above_vwap()`	当前价 > VWAP
动量门控	`_gate_momentum()`	最新价 > 3根 Bar 前价格
量能门控	`_gate_volume_surge()`	成交量 ≥ 1.5 × 近期均量
波动率门控	`_gate_realized_volatility()`	实现波动率 < 4%

可配置参数：

参数	默认值	说明
min_offline_score	70	最低离线评分阈值
min_intraday_score	70	最低盘中评分阈值
default_stop_pct	0.04	默认止损比例（4%）
default_reward_risk_ratio	2.0	默认盈亏比
volume_surge_threshold	1.5	量能突破倍数
max_realized_volatility	0.04	最大允许实现波动率（4%）

核心函数：

evaluate_entry_signal() — 主评估函数
compute_intraday_score() — 计算 0-100 盘中评分
build_trade_plan() — 根据配置计算止损/目标价

backtest_api.py — 因子回测 API

因子评估和策略绩效的 REST API。

端点	方法	说明
`/api/backtest/factors`	GET	因子评估（IC/ICIR 统计、IC 分布）
`/api/backtest/performance`	GET	策略绩效（NAV、Sharpe、回撤、月度收益）
`/api/backtest/compare`	GET	多策略对比
`/api/backtest/cache/clear`	POST	清除计算缓存

预定义因子：momentum_20d, momentum_5d, rsi_14, volume_ratio_5

绩效指标：NAV 曲线、Sharpe/Calmar 比率、最大回撤、年化收益/波动率、胜率、超额收益

presets.py — 多时间周期预设策略

定义跨日/周/60 分钟的多时间周期共振策略。

4 种预设策略：

预设	说明
trend_following	趋势追踪（多周期均线共振）
pullback_buy	回调买入（趋势确认 + 回踩支撑）
momentum_breakout	动量突破（多周期突破共振）
value_quality	价值质量（基本面 + 技术面共振）

每个预设在日/周/60 分钟三个周期分别定义 conditions，后端根据快照数据计算匹配度并返回评分。

market_hours.py — 交易时段

美股交易时段感知工具。

功能：

判断当前交易时段（盘前/盘中/盘后/空闲）
美东时区处理（America/New_York）
内置假日日历（2026-2027 年）
决定实时数据获取策略（空闲时段暂停）

rt_api.py — 实时数据路由

提供盘中实时数据的 API 端点。

端点	说明	可用性
`/api/rt/quotes`	获取实时报价	免费
`/api/rt/ranking`	实时排名	付费
`/api/rt/kline`	分钟 K 线	付费
`/api/rt/scan`	全市场扫描	付费
`/api/rt/status`	服务状态查询	免费
`/api/rt-prices`	轻量价格轮询端点	免费

rt_hub.py — 实时推送中心

实时报价的核心编排模块，连接上游数据源和下游 WebSocket 客户端。

功能：

管理所有 WebSocket 客户端连接
处理订阅/取消订阅（每个客户端可订阅不同股票）
接收上游推送并广播给订阅的客户端
交易时段感知（非交易时段自动暂停推送）

rt_ws.py — WebSocket 端点

提供 /ws/rt WebSocket 端点供前端连接。

功能：

设备注册与断线清理
subscribe/unsubscribe 消息处理
ping/pong 心跳检测
连接状态追踪

rt_aliyun_ticker.py — 阿里云 Ticker

后台定时任务，通过阿里云 API 获取实时行情并合并到快照缓存。

特性：

可配置更新间隔（默认 60 秒）
交易时段感知（盘前显示 after_price，盘中显示 price，盘后显示 after_price）
支持启动/停止控制
通过 config.yaml 中的 aliyun_ticker_enabled 开关

alerts_api.py — 预警 API

预警系统的 REST API 接口。

端点	方法	说明
`/api/alerts/rules`	GET	获取预警规则列表
`/api/alerts/rules`	POST	创建预警规则
`/api/alerts/rules/{id}`	DELETE	删除预警规则
`/api/alerts/events`	GET	获取触发事件历史（分页）
`/api/alerts/presets`	GET	获取预警模板列表

参数验证：每设备最多 50 条规则，规则类型和参数校验。

alerts_db.py — 预警 SQLite 存储

预警系统的持久化存储层。

数据表：

表	用途
devices	设备注册信息
alert_rules	预警规则（类型、参数、状态）
alert_events	触发事件记录

特性：

WAL 模式支持并发读写
事件自动清理（默认保留 30 天）
外键级联删除

alert_engine.py — 预警引擎

核心预警处理模块，负责规则评估、状态管理和事件分发。

工作流：

实时报价到达
  ↓
遍历所有 armed 状态的规则
  ↓
alert_evaluators 评估条件
  ↓
条件满足 → 触发事件
  ├── 状态变为 fired
  ├── 写入 SQLite (alert_events)
  └── alert_dispatcher → WebSocket 推送到前端

alert_evaluators.py — 规则评估器

6 种预警类型的条件评估逻辑。

评估器	类型	说明
价格变动	price_change	涨跌幅超过阈值（如 ±3%）
突破新高	breakout_high	价格突破指定高点
跌破新低	breakout_low	价格跌破指定低点
成交量异动	volume_spike	成交量超过 N 倍均值
均线交叉	ma_cross	短期均线穿越长期均线
52 周高低	week52_high_low	接近或突破 52 周高低点

alert_models.py — 数据结构

预警系统的类型定义。

AlertRule — 用户创建的预警规则
CompiledRule — 编译后的规则（含评估器）
AlertEvent — 触发的事件记录
AlertPreset — 预警模板

alerts_presets.py — 预警模板

6 种预定义预警模板，方便用户快速创建规则：

模板	说明
急涨急跌	涨跌幅超过阈值
突破新高	价格突破近期高点
跌破新低	价格跌破近期低点
放量异动	成交量异常放大
均线金叉	短期均线上穿长期均线
52 周新高	接近 52 周最高价

alert_dispatcher.py — 预警分发器

将触发的预警事件通过 WebSocket 推送到对应设备的前端。

维护设备 ID → WebSocket 连接的映射
Best-effort 交付（连接断开则跳过）
与 AlertEngine 事件回调对接

screener_api.py — 选股器 API

提供选股结果和评分排名接口。

端点	方法	说明
`/api/screener/status`	GET	选股器运行状态
`/api/screener/results`	GET	排名结果列表
`/api/screener/stock/{symbol}`	GET	单只股票评分详情

macro.py — 宏观指标接口

读取 macro.parquet 并提供宏观指标 API。

端点	方法	说明
`/api/macro`	GET	获取宏观指标数据

返回内容：

各指标（VIX、10Y、DXY、SPX）的最新值和变化
风险等级分类：calm（平静）/ normal（正常）/ risk（风险）
缓存机制（mtime 感知，文件更新自动刷新）

sectors.py — 板块轮动接口

读取 sectors.parquet 并提供板块轮动数据 API。

端点	方法	说明
`/api/sectors`	GET	获取板块轮动数据

返回内容：

11 个板块 ETF + 3 个风格 ETF 的排名
20 日 / 60 日收益率
相对 SPY 基准的相对强弱
动量评分

prod_app.py — 生产环境包装

在主应用基础上增加静态文件服务。如果 static/ 目录存在，自动挂载为静态资源路由。

运行命令

cd backend
source .venv/bin/activate
uvicorn main:app --reload --port 8000

美股-数据管线模块

Henry Lin — Sun, 14 Jun 2026 09:19:26 +0000

数据管线模块（scripts/）

数据管线负责从外部数据源获取股票行情和财务数据，计算技术指标和基本面评分，并构建离线快照文件供后端服务使用。同时包含宏观指标、板块数据、信号候选池和信号绩效的构建脚本。

目录结构

scripts/
├── config.yaml            # 股票池与获取配置
├── yfinance_client.py     # yfinance API 封装
├── feature_engine.py      # 技术指标与形态识别
├── scorer.py              # 综合评分 + 价值-成长复合评分
├── build_snapshot.py      # 快照构建入口（编排器）
├── build_macro.py         # 宏观指标数据构建
├── build_sectors.py       # 板块 ETF 数据构建
├── rt_manager.py          # 实时数据管理器
├── rt_providers.py        # 实时数据源抽象层（yfinance / 阿里云）
├── rt_push_providers.py   # 实时推送源实现（TickFlow / Ashen / 阿里云）
├── ashen_ws_client.py     # Ashen WebSocket 客户端
├── signal_candidate_pool.py # 信号候选池离线打分
├── signal_outcomes.py     # 信号绩效计算（MFE/MAE/前向收益）
├── dev_server.sh          # 开发服务器管理脚本
└── daily5_alpha_select.py # Alpha 因子选股脚本

数据流

config.yaml
    ↓
yfinance_client.py  ──获取──→  原始行情 + 股票信息 + 财务指标
    ↓
feature_engine.py   ──计算──→  技术指标（~80 列）+ 高级指标（MACD/RSI/BB/VWAP/MFI/Williams %R/ADX/CCI/Stochastic/OBV/Chaikin/ATR）
    ↓
scorer.py           ──评分──→  signal_score（0-100）+ value_quality_score + piotroski_score
    ↓
build_snapshot.py   ──输出──→  data.parquet + history.parquet

build_macro.py      ──获取──→  macro.parquet（VIX / 10Y / DXY / SPX）
build_sectors.py    ──获取──→  sectors.parquet（14 个板块/风格 ETF）

signal_candidate_pool.py ──打分──→  candidates.parquet / candidates.csv（每日 Top 20）
signal_outcomes.py       ──计算──→  MFE/MAE + 1/3/5/10 日前向收益

各文件详解

config.yaml — 配置中心

定义股票池、数据获取参数、实时数据配置和预警配置。

股票池（约 200 只，分 9 组）：

分组	说明	示例
tech	科技/半导体（70+只）	AAPL, NVDA, MSFT, TSLA, ARM, TSM
finance	金融/支付	JPM, V, MA, BRK-B, COIN
healthcare	医疗/生物科技	JNJ, UNH, LLY, MRNA
energy	能源/资源	XOM, CVX, COP, NEM, FCX
industrial	工业/航空/国防	CAT, BA, GE, HON, RTX
consumer	消费/零售/餐饮	PG, KO, MCD, WMT, NFLX
infra	通信/公用事业/REIT	T, VZ, NEE, AMT, EQIX
aerospace	太空/航空/国防	RKLB, ASTS, LUNR, LHX, HEI
other	汽车/交通/其他	F, RIVN, UBER, BIDU, PDD

获取参数：

参数	默认值	说明
history_period	2y	历史数据时间跨度
batch_size	20	每批获取股票数量
batch_delay	2.0s	批次间隔
max_retries	2	最大重试次数
request_timeout	30s	单次请求超时

多时间周期配置：

时间周期	interval	period
日线	1d	2y
周线	1wk	5y
60 分钟	60m	730d
30 分钟	30m	60d
15 分钟	15m	60d
5 分钟	5m	60d

实时数据配置：

配置	说明
`realtime.provider`	数据源选择：free (yfinance) / premium (aliyun)
`realtime.push.provider`	推送源选择：tickflow / ashen / aliyun
`realtime.trading_hours`	交易时段定义（盘前/盘中/盘后）
`realtime.update_interval_sec`	报价更新间隔（秒）
`alerts.enabled`	预警引擎开关
`alerts.db_path`	SQLite 数据库路径
`alerts.max_rules_per_device`	每设备最大规则数（50）
`alerts.retention_days`	事件保留天数（30）

yfinance_client.py — 数据获取

封装 yfinance API，提供批量获取、重试和标准化功能。

核心函数：

函数	用途	返回
`load_config()`	加载 config.yaml	dict
`get_stock_pool()`	提取去重后的股票列表	list[str]
`fetch_history_batch()`	批量下载历史行情	dict[str, DataFrame]
`fetch_stock_info()`	获取单只股票基本信息	StockInfo
`fetch_financial_metrics()`	获取财务指标	dict
`fetch_all_data()`	主入口，获取全部数据	(history, info, financials)

数据类：

FetchConfig — 获取参数（history_period, batch_size 等）
StockInfo — 股票基本信息（symbol, name, market_cap, total_share）

处理逻辑：

分批下载避免 API 限流
自动重试失败请求
列名标准化（yfinance 格式 → 统一格式）

feature_engine.py — 技术指标计算

从原始 OHLCV 数据中计算约 80 个技术指标列。

指标分类：

移动平均与趋势

指标	说明
ma5, ma10, ma20, ma60	简单移动平均线
ret_1d, ret_5d, ret_20d	收益率
amplitude	振幅

K 线形态识别（二值标志列，0/1）

单 K 形态：bull_candle, bear_candle, hammer, shooting_star, doji, dragonfly_doji, gravestone_doji

双 K 形态：bullish_engulfing, bearish_engulfing, inside_bar

三 K 形态：three_white_soldiers, three_black_crows

趋势形态：ma_bull, trend_confirm, pullback_hold, break_high_20, break_high_60, platform_breakout, false_breakout, double_bottom

量价形态：volume_breakout, volume_shrink_pullback

波动率形态：volatility_contraction, volatility_expansion

连续上涨：consec_up_3, consec_up_5

高级技术指标（连续值 + 信号标志）

MACD（移动平均收敛发散）

列名	类型	说明
macd_dif	连续值	DIF 线（12日 EMA - 26日 EMA）
macd_dea	连续值	DEA 信号线（DIF 的 9日 EMA）
macd_hist	连续值	MACD 柱状图 = 2 × (DIF - DEA)
ind_macd_bull_cross	标志	金叉（DIF 上穿 DEA）
ind_macd_bear_cross	标志	死叉（DIF 下穿 DEA）
ind_macd_hist_positive	标志	柱状图由负转正

RSI（相对强弱指数）

列名	类型	说明
rsi_14	连续值	14日 RSI（Wilder 平滑）
ind_rsi_oversold	标志	RSI < 30（超卖）
ind_rsi_overbought	标志	RSI > 70（超买）
ind_rsi_bull_divergence	标志	价格新低但 RSI 未新低（看涨背离）

Bollinger Bands（布林带）

列名	类型	说明
bb_upper	连续值	上轨 = MA20 + 2 × 标准差
bb_mid	连续值	中轨 = 20日 SMA
bb_lower	连续值	下轨 = MA20 - 2 × 标准差
bb_width	连续值	带宽百分比 = (上轨 - 下轨) / 中轨 × 100
ind_bb_squeeze	标志	带宽处于 20日最低值（即将突破）
ind_bb_upper_touch	标志	价格触及或超过上轨
ind_bb_lower_touch	标志	价格触及或跌破下轨

VWAP（成交量加权平均价）

列名	类型	说明
vwap	连续值	当日成交量加权均价（日内累计，每个交易日重置）
ind_price_above_vwap	标志	收盘价 > VWAP

MFI（资金流量指数）

列名	类型	说明
mfi_14	连续值	14日 MFI（带量的 RSI）
ind_mfi_oversold	标志	MFI < 20（超卖）
ind_mfi_overbought	标志	MFI > 80（超买）

Williams %R（威廉指标）

列名	类型	说明
willr_14	连续值	14日 Williams %R（范围 -100 ~ 0）
ind_willr_oversold	标志	Williams %R < -80（超卖）
ind_willr_overbought	标志	Williams %R > -20（超买）

ADX（平均趋向指数）

列名	类型	说明
adx_14	连续值	14日 ADX（趋势强度）
di_plus_14	连续值	+DI（正向趋向指标）
di_minus_14	连续值	-DI（负向趋向指标）
ind_adx_strong_trend	标志	ADX > 25（强趋势）
ind_di_bull_cross	标志	+DI 上穿 -DI

CCI（商品通道指数）

列名	类型	说明
cci_20	连续值	20日 CCI
ind_cci_oversold	标志	CCI < -100（超卖）
ind_cci_overbought	标志	CCI > 100（超买）

Stochastic（随机指标）

列名	类型	说明
stoch_k_14	连续值	14日 %K（快速）
stoch_d_14	连续值	%K 的 3日 SMA（慢速）
ind_stoch_oversold	标志	%K < 20（超卖）
ind_stoch_overbought	标志	%K > 80（超买）
ind_stoch_bull_cross	标志	%K 上穿 %D（且 %K < 80）

OBV（能量潮）

列名	类型	说明
obv	连续值	累计能量潮
ind_obv_uptrend	标志	OBV > 5日前 OBV

Chaikin Oscillator（佳庆振荡器）

列名	类型	说明
chaikin_osc	连续值	A/D 线的 3日 EMA - 10日 EMA
ind_chaikin_bullish	标志	Chaikin 振荡器上穿零线

ATR（平均真实范围）

列名	类型	说明
atr_14	连续值	14日 ATR（Wilder 平滑）
atr_stop_long	连续值	Chandelier 止损 = 周期最高价 - 3 × ATR

成交量与波动率

指标	说明
vol_ratio_5	5 日量比
body_ratio	实体比率
upper/lower_shadow_ratio	上下影线比率
risk_ma20_dev, risk_ma60_dev	均线偏离度
risk_dd_20d	20 日最大回撤
risk_vol_20d	20 日波动率

基本面评分指标

列名	类型	说明
piotroski_score	连续值	Piotroski F-Score (0-9)
fund_piotroski_strong	标志	Piotroski ≥ 7（财务强健）
altman_z_score	连续值	Altman Z-Score（破产预警）
risk_financial_distress	标志	Altman Z < 1.81（财务困境）
value_quality_score	连续值	价值-成长复合评分 (0-100)
fund_value_quality	标志	复合评分 ≥ 70

核心函数：

add_features(df) — 主入口，给历史 DataFrame 添加全部技术指标
add_financial_features(df, metrics) — 将财务指标合并到最新行（含 Piotroski 评分）

scorer.py — 综合评分

计算每只股票的信号评分（0-100 分）和价值-成长复合评分。

信号评分组成：

维度	说明	分值范围
标志指标	多头信号加分，空头信号减分	-20 ~ +50
连续指标	收益率、ROE、利润率等百分位排名	0 ~ +25
PE 惩罚	forward_pe 过高时扣分	-10 ~ 0

关键权重：

信号	权重
pattern_ma_bull	+15
trend_confirm	+12
break_high_60	+10
false_breakout	-8
three_black_crows	-6

价值-成长复合评分（compute_value_quality_score()）：

维度	权重	排名方向	包含指标
估值质量	40%	反向（低估值 = 高分）	forward_pe, price_to_book, peg_ratio, ps_ratio
盈利能力	30%	正向	roe, roa, profit_margin, operating_margin
成长性	30%	正向	revenue_growth, earnings_growth

build_snapshot.py — 快照编排器

协调数据获取 → 特征计算 → 基本面评分 → Altman Z-Score → 输出的完整流程。

核心函数：

函数	用途
`build_snapshot()`	主入口，输出 parquet 文件
`build_snapshot_dataframe()`	构建快照 DataFrame（每只股票一行）
`build_history_dataframe()`	构建历史 DataFrame（全部日期）
`_compute_altman_z(balance_sheet)`	计算 Altman Z-Score（破产预警）

Altman Z-Score 公式：

Z = 1.2×X1 + 1.4×X2 + 3.3×X3 + 0.6×X4 + 1.0×X5

变量	定义
X1	营运资本 / 总资产
X2	留存收益 / 总资产
X3	EBIT / 总资产
X4	市值 / 总负债
X5	营收 / 总资产

Piotroski F-Score（在 add_financial_features() 中调用 _add_piotroski_score()）：

9 项评分标准：

#	维度	条件
1	盈利能力	ROA > 0
2	现金流	经营利润率 > 0（作为经营现金流代理）
3	ROA 改善	盈利增长 > 0
4	应计项目	经营利润率 > ROA
5	杠杆	负债权益比 < 1.0
6	流动性	流动比率 > 1.5
7	股本稀释	ROA > 0.05（代理）
8	利润率	净利润率 > 0.15
9	营收增长	营收增长 > 0

输出文件：

data.parquet — 最新交易日的快照（~200 行 × 80+ 列）
history.parquet — 全部历史技术指标数据

build_macro.py — 宏观指标构建

获取宏观经济指标历史数据，输出 macro.parquet。

跟踪指标：

Symbol	指标	说明
^VIX	VIX 恐慌指数	市场波动率预期
^TNX	10 年期美债收益率	无风险利率基准
DX-Y.NYB	美元指数 DXY	美元强弱
^GSPC	标普 500 指数	大盘基准

处理逻辑：

10Y 收益率自动转换为百分比格式
每个指标独立获取，单个失败不影响其他
输出格式：date, symbol, close, change_pct

build_sectors.py — 板块数据构建

获取板块 ETF 和风格 ETF 的历史数据，输出 sectors.parquet。

11 个板块 ETF：XLK, XLF, XLV, XLE, XLI, XLY, XLU, XLRE, XLB, XLC, IBB

3 个风格 ETF：SPY（大盘基准）, QQQ（科技基准）, IWM（小盘基准）

输出格式：date, symbol, close, volume

signal_candidate_pool.py — 信号候选池

离线多因子打分脚本，输出每日信号候选股 Top 20。

评分维度与权重：

维度	权重	包含因子
基本面	0.25	盈利增长、ROE、利润率、PE 惩罚
周线	0.20	20日动量、signal_score、MA 排列
日频 Alpha	0.30	signal_score、量能突破、突破信号
技术面	0.25	ADX 趋势、MA 排列、风险指标

风险门控：

低流动性：20日日均成交额 < $5M → 标记
高波动率：20日波动率 > 4.0% → 标记
周线下跌趋势 → 标记

CLI 用法：

python3 scripts/signal_candidate_pool.py \
    --snapshot data.parquet \
    --output tmp/signals/candidates.parquet \
    --csv tmp/signals/candidates.csv

signal_outcomes.py — 信号绩效计算

计算已入场信号的后验绩效指标。

产出指标：

指标	说明
forward_return_1d	1 日前向收益
forward_return_3d	3 日前向收益
forward_return_5d	5 日前向收益
forward_return_10d	10 日前向收益
MFE	最大有利偏移（Maximum Favorable Excursion）
MAE	最大不利偏移（Maximum Adverse Excursion）
hit_stop	是否触及止损
hit_target	是否触及目标价

CLI 用法：

python3 scripts/signal_outcomes.py \
    --db tmp/signals/signals.db \
    --history history.parquet

rt_manager.py & rt_providers.py — 实时数据

提供盘中实时数据的统一接口。

rt_providers.py 定义两种数据源：

数据源	延迟	排名	分钟 K 线
YFinanceProvider（免费）	~15 分钟	不支持	不支持
AliyunProvider（付费）	实时	支持	支持

rt_manager.py 提供缓存层：

LRU 缓存（报价 200 条，K 线 50 条）
统一的异步 API 接口

rt_push_providers.py — 实时推送源

可插拔的实时推送上游数据源架构，支持多种供应商。

支持的供应商：

供应商	协议	说明
TickFlowProvider	HTTP 轮询	默认供应商，API Key 认证
AshenProvider	WebSocket	增量订阅，自动重连
AliyunProvider	HTTP 轮询	阿里云金融 API

特性：

统一的 Provider 接口（工厂模式）
可配置轮询间隔和推送频率
自动重连与退避策略

ashen_ws_client.py — Ashen WebSocket 客户端

Ashen 实时行情服务的 WebSocket 客户端实现。

增量订阅管理（add/remove symbols）
自动重连（指数退避：1s → 2s → 5s → 10s）
心跳检测与超时处理

运行命令

# 构建数据快照
python3 scripts/build_snapshot.py

# 构建宏观指标
python3 scripts/build_macro.py

# 构建板块数据
python3 scripts/build_sectors.py

# 生成信号候选池
python3 scripts/signal_candidate_pool.py --snapshot data.parquet --output tmp/signals/candidates.parquet

# 计算信号绩效
python3 scripts/signal_outcomes.py --db tmp/signals/signals.db --history history.parquet

美股日线选股器 — 项目概览

Henry Lin — Sun, 14 Jun 2026 09:19:01 +0000

美股日线选股器 — 项目概览

项目简介

美股日线选股器（US Stock Screener）是一个基于技术形态和价值指标的股票筛选工具。系统从 yfinance 获取约 200 只美股的日线数据，计算 80+ 个技术指标和形态信号，通过 FastAPI 后端和 React 前端提供交互式筛选、排序和对比功能。

系统还集成了实时行情推送（WebSocket）、预警系统（6种规则类型）、宏观指标监控、板块轮动分析、多时间周期共振策略、基本面评分体系、信号产品系统（候选池→盘中信号→模拟交易→绩效分析）和因子回测等高级功能。

技术栈

层级	技术	说明
数据获取	yfinance	免费 API，无需密钥
特征工程	pandas + numpy	技术指标计算
后端	FastAPI + uvicorn	异步 API 服务
前端	Vite + React 18 + TypeScript	SPA 单页应用
数据存储	Parquet 文件	列式存储，高效压缩
信号存储	SQLite	候选池、信号、模拟仓位持久化
预警存储	SQLite	规则与事件持久化
实时数据	yfinance / 阿里云 / TickFlow	免费（延迟）/ 付费（实时）
实时推送	WebSocket	服务端主动推送报价
K线图表	lightweight-charts	交互式技术图表
路由	react-router-dom	SPA 页面路由

架构总览

┌──────────────────────────────────────────────────────────────────┐
│                      Frontend (React 18 SPA)                        │
│  筛选面板 │ 股票表格 │ 详情抽屉 │ K线图 │ 对比分析 │ 预警中心      │
│  宏观指标 │ 板块轮动 │ 多周期预设 │ 数值筛选 │ 实时扫描器            │
│  信号中心 │ 信号绩效 │ 模拟组合 │ 回测报告 │ 形态统计               │
└─────────────────────────────┬────────────────────────────────────┘
                              │ HTTP + WebSocket (proxy :5173 → :8000)
┌─────────────────────────────▼────────────────────────────────────┐
│                    Backend (FastAPI + uvicorn)                       │
│  /api/meta │ /api/stocks │ /api/rt/* │ /api/alerts │ /api/macro     │
│  /api/sectors │ /api/pattern-stats │ /api/presets │ /ws/rt          │
│  /api/signals/* │ /api/backtest/*                                   │
│  loader.py (内存缓存) ← data.parquet (200+行 × 80+列)               │
│  history.parquet │ fundamentals.parquet │ macro/sectors parquet       │
│  RTHub (推送中心) │ AlertEngine (预警引擎) │ SQLite (规则/信号存储)    │
│  SignalEntryEngine (信号入场引擎) │ SignalProductRepository           │
└─────────────────────────────┬────────────────────────────────────┘
                              ↑ 构建时生成
┌─────────────────────────────┴────────────────────────────────────┐
│                  Data Pipeline (scripts/)                            │
│  yfinance_client → feature_engine → scorer → parquet output         │
│  rt_manager → rt_providers → rt_push_providers (实时数据)            │
│  build_macro → macro.parquet │ build_sectors → sectors.parquet      │
│  signal_candidate_pool → candidates │ signal_outcomes → MFE/MAE     │
└──────────────────────────────────────────────────────────────────┘

模块说明

模块	文档	说明
scripts/	01-data-pipeline.md	数据获取、特征工程、快照构建、宏观/板块数据、信号候选池
backend/	02-backend-api.md	API 服务、查询引擎、实时推送、预警系统、信号产品、回测接口
frontend/	03-frontend-ui.md	筛选界面、详情、图表、对比、预警、信号中心、回测报告
demo/	04-factor-backtest.md	Alpha101 因子回测、中性化、多空选股、多信号共振
scripts/screener/	05-realtime-screener.md	多因子实时选股引擎（基本面/周线/Alpha/盘中）
—	06-deployment.md	部署、开发服务器、测试体系、配置管理

核心功能

1. 技术形态筛选

支持 30+ 种 K 线形态和技术信号的组合筛选，包括单 K 形态、双 K 形态、三 K 形态、趋势结构、量价关系和波动率形态。

2. 高级技术指标

80+ 个技术指标覆盖 6 大类别：

类别	指标	信号列
趋势	MACD (DIF/DEA/Histogram)、ADX (+DI/-DI)	金叉/死叉、强趋势
动量	RSI-14、Stochastic (%K/%D)、CCI-20、Williams %R	超买/超卖、背离
波动率	Bollinger Bands (上/中/下/宽度)、ATR + 止损	挤压信号、触及上下轨
量能	OBV、Chaikin Oscillator、MFI-14、VWAP	量价背离、资金进出
财务健康	Piotroski F-Score (0-9)、Altman Z-Score	财务强健/财务困境
综合评分	Value-Growth Composite (0-100)	价值质量

3. 基本面评分体系

三维度基本面分析框架：

Piotroski F-Score：9 项财务健康指标评分（0-9 分），≥7 分标记为"财务强健"
Altman Z-Score：破产预警模型，<1.81 标记为"财务困境"
价值-成长复合评分：估值质量 (40%) + 盈利能力 (30%) + 成长性 (30%)

4. 信号产品系统

完整的交易信号管线，从离线候选到盘中信号到绩效追踪：

候选池生成：离线多因子打分，输出每日 Top 20 候选股
盘中信号入场：4 重入场门控（VWAP、动量、量能、波动率）
模拟交易：Paper Trading 跟踪开仓/平仓/盈亏
绩效分析：MFE/MAE、多周期前向收益 (1/3/5/10 日)、胜率

5. 综合评分

基于多维度的信号评分系统（0-100 分），综合考虑技术形态、连续指标和估值水平。

6. 股票详情

详情抽屉展示完整的股票信息，包含基本面、技术指标、财务数据和风险指标。研究页面提供 K 线图、技术面分析、财务面分析和形态历史回溯。

7. 多股对比

最多 5 只股票的并列对比分析，按 7 个维度比较并高亮最优/最差值。

8. 实时行情

盘中实时数据支持：

免费源：yfinance（~15 分钟延迟）
付费源：阿里云金融 API（真正实时）
WebSocket 推送：服务端主动推送报价到前端
实时排名：按成交额/涨跌幅/成交量排序
异动扫描：涨幅榜、跌幅榜、成交量异动

9. 预设策略

内置 5 种选股策略（趋势突破、回调低吸、质量成长、估值修复、风险预警），一键应用筛选组合。

10. 多时间周期共振

支持日线、周线、60 分钟、30 分钟、15 分钟、5 分钟多周期分析，提供跨周期预设策略与共振评分。

11. 预警系统

6 种预警规则类型（价格变动、突破新高/新低、成交量异动、均线交叉、52 周高低），通过 WebSocket 实时推送 Toast 通知。

12. 宏观指标监控

实时展示 VIX 恐慌指数、10 年期美债收益率、美元指数 DXY、标普 500 等宏观指标，自动计算风险等级（平静/正常/风险）。

13. 板块轮动分析

11 个板块 ETF + 3 个风格 ETF 的相对强弱排名，展示动量和相对 SPY 基准的表现。

14. 因子回测与回测报告

Alpha101 因子评估和多因子选股回测系统：

因子 IC/ICIR 评估与中性化
多空组合绩效分析（Sharpe、回撤、Calmar）
多信号共振回测（形态组合胜率统计）
交易成本建模（佣金、滑点、融券成本）
前端可视化回测报告页面

快速启动

# 1. 构建数据快照
python3 scripts/build_snapshot.py

# 2. 构建宏观与板块数据
python3 scripts/build_macro.py
python3 scripts/build_sectors.py

# 3. 启动后端
cd backend && source .venv/bin/activate
uvicorn main:app --reload --port 8000

# 4. 启动前端
cd frontend && npm install && npm run dev

# 或使用统一脚本
bash scripts/dev_server.sh start

数据说明

股票池：约 200 只美股，覆盖科技、金融、医疗、能源、工业、消费、基建、太空、汽车等 9 个板块
数据源：yfinance（免费，无需 API Key），可选阿里云（付费，实时）
更新频率：手动运行 build_snapshot.py 更新快照
技术指标：约 80 个，包含移动平均、K 线形态、量价关系、波动率、动量、趋势等
财务指标：PE、PEG、ROE、利润率、营收增长、Piotroski F-Score、Altman Z-Score 等
实时推送：WebSocket 服务端推送，支持 TickFlow / Ashen / 阿里云三种上游源
预警规则：SQLite 存储，每设备最多 50 条规则，30 天事件保留
信号系统：SQLite 持久化候选池、入场信号、模拟仓位和绩效追踪

开源量化分析软件功能介绍

Henry Lin — Mon, 08 Jun 2026 06:37:06 +0000

📈 多因子选股系统 — 项目功能介绍

观看地址:https://youtu.be/SpHsZdlyii8

面向 A 股市场的全链路量化分析平台 — 因子计算 · 机器学习 · 组合优化 · 回测验证 · 实时行情

指标	数据
Python 模块	108
业务代码	15,600+ 行
服务模块	30+
API 蓝图	14
内置因子	12

一、整体功能介绍

多因子选股系统是一个覆盖量化研究全流程的分析平台，从数据管理到策略回测一站式完成。系统采用 Flask + SQLAlchemy + SocketIO 技术栈，基于 Parquet + SQLite 双层存储架构，零外部数据库依赖，克隆即可运行。

核心能力总览

实时行情分析：通达信分钟数据接入，MACD、KDJ、RSI、布林带等技术指标实时计算
因子与选股：12 个内置因子 + 自定义因子表达式引擎，支持白名单安全校验
机器学习建模：XGBoost、LightGBM、RandomForest 三大算法的完整模型生命周期
组合优化：等权重、均值方差、风险平价、因子中性四种经典优化方法
回测验证：单策略与多策略回测，丰富的绩效指标（年化收益、夏普比率、最大回撤等）
风险管理：组合持仓管理、实时价格刷新、风险指标监控、预警管理、压力测试

请求流程

HTTP / WebSocket → Blueprint (app/api/*.py) → Service (app/services/*.py) → ParquetDataReader (行情数据)
                                                                             ↕
                                                               SQLAlchemy Models (SQLite 应用状态)

技术栈

类别	技术
后端框架	Python 3.8–3.11、Flask、Flask-SocketIO、Flask-SQLAlchemy、Celery
数据处理	Pandas、NumPy、PyArrow、SciPy、Scikit-learn
机器学习	XGBoost、LightGBM、RandomForest、CVXPY
前端 & 可视化	Bootstrap 5、ECharts、Plotly、Matplotlib / MPLFinance
数据源	通达信（实时行情）、Baostock（历史数据）、Tushare（可选）、Parquet 离线数据包
AI 能力	Ollama（本地 LLM）、OpenAI API（可选）、Text2SQL、Qwen2.5-Coder

二、各模块功能介绍

2.1 实时行情分析模块

API 蓝图前缀：/api/realtime-analysis/*

核心服务：

RealtimeIndicatorEngine（services/realtime_indicator_engine.py）：实时技术指标计算
RealtimeTradingSignalEngine（services/realtime_trading_signal_engine.py）：多策略交易信号生成与融合

功能特性：

功能	说明
实时指标计算	支持 MACD、KDJ、RSI、布林带等经典技术指标的实时计算与展示
多策略信号生成	基于不同策略的交易信号生成，支持信号融合与综合研判
信号监控	交易信号的实时监控与历史回溯
策略回测	对生成的信号进行历史回测验证
板块分析	板块表现统计与异动检测
市场情绪	市场情绪指标的实时追踪

数据源：通达信分钟级行情数据，存储于 data/stock_minute/（Parquet 格式）。

2.2 因子与选股模块

API 蓝图前缀：/api/ml-factor/*、/api

核心服务：

FactorEngine（services/factor_engine.py）：内置因子计算与自定义因子管理
FactorExpressionEngine（services/factor_expression_engine.py）：白名单校验的自定义因子公式引擎
StockScoringEngine（services/stock_scoring.py）：因子评分与 ML 评分的综合选股引擎

内置因子（12 个）

维度	因子名	说明
动量	`momentum_1d`	1 日动量
动量	`momentum_5d`	5 日动量
动量	`momentum_20d`	20 日动量
波动率	`volatility_20d`	20 日波动率
技术指标	`rsi_14`	RSI 相对强弱
技术指标	`turnover_rate`	换手率
基本面	`pe_ratio`	市盈率
基本面	`pb_ratio`	市净率
基本面	`roe`	净资产收益率
基本面	`debt_ratio`	资产负债率
基本面	`current_ratio`	流动比率
基本面	`gross_margin`	毛利率

自定义因子

支持用户通过表达式语法定义自定义因子，表达式引擎采用白名单校验机制，确保安全性。用户可在因子管理界面创建、编辑和管理自定义因子公式。

选股评分

StockScoringEngine 提供两种评分模式：

因子评分：基于多个因子的加权综合评分
ML 评分：基于机器学习模型的预测评分

2.3 机器学习建模模块

API 蓝图前缀：/api/ml-factor/*

核心服务：

MLModelManager（services/ml_models.py）：XGBoost / LightGBM / RandomForest 模型生命周期管理
ModelTrainingJobService（services/model_training_job_service.py）：异步训练任务轮询与管理

功能特性：

功能	说明
模型创建	支持 XGBoost、LightGBM、RandomForest 三大算法的模型创建
模型训练	支持同步与异步训练，异步训练通过 Celery 或内联模式执行
模型预测	训练完成的模型可直接用于股票预测
特征工程	因子组合与特征选择
性能评估	模型性能评估与多模型比较
训练任务管理	异步训练任务的提交、状态查询与管理

2.4 组合优化模块

API 蓝图前缀：/api/ml-factor/*

核心服务：

PortfolioOptimizer（services/portfolio_optimizer.py）：等权 / 均值方差 / 风险平价 / 因子中性

优化方法：

方法	说明	适用场景
等权重优化	所有股票分配相同权重	简单基准、等权组合
均值方差优化	基于 Markowitz 理论的均值-方差优化	追求风险调整后收益最大化
风险平价优化	基于风险贡献均衡的权重分配	风险均衡配置
因子中性组合	控制因子暴露度在中性水平	消除特定因子风险暴露

2.5 回测验证模块

API 蓝图前缀：/api/ml-factor/*

核心服务：

BacktestEngine（services/backtest_engine.py）：单策略与多策略回测

回测能力：

功能	说明
单策略回测	对单个选股策略进行历史回测验证
多策略比较	同时运行多个策略并比较绩效表现
年化收益率	计算策略的年化收益
夏普比率	风险调整后收益指标
最大回撤	策略的最大净值回撤幅度
胜率	盈利交易占比
卡尔玛比率	年化收益与最大回撤的比值

回测结果以可视化图表形式展示，支持净值曲线、收益分布等多种视图。

2.6 风险管理模块

API 蓝图前缀：/api/realtime-analysis/*

核心服务：

RealtimeRiskManager（services/realtime_risk_manager.py）：风险指标、预警管理、压力测试

功能特性：

功能	说明
投资组合持仓 CRUD	创建、查询、更新、删除投资组合持仓记录
实时价格刷新	通过通达信接口获取实时行情价格
风险指标监控	实时计算并展示组合风险指标
预警管理	设定预警规则，触发时自动通知
压力测试	模拟极端市场场景，分析组合表现

2.7 数据管理模块

API 蓝图前缀：/api/data-jobs/*

核心服务：

DataJobs（services/data_jobs/）：数据下载任务执行器

数据架构：

存储	用途	说明
Parquet 文件	市场行情数据	高性能列式存储，承载日线/分钟线行情、基本面、技术面、资金流数据
SQLite 数据库	应用状态	轻量级存储，管理持仓、报告、预警等低并发元数据
Redis 缓存	任务队列 & 缓存	Celery 任务 Broker、实时数据缓存、WebSocket 消息分发（可选）

数据下载工具（位于 app/utils/）：

工具	数据源	说明
`trade_calendar.py`	Tushare / Baostock	交易日历下载
`stock_basic.py`	Tushare / Baostock	股票基本信息下载
其他数据脚本	Tushare / Baostock	行情、财务等数据下载

注意：下载顺序为 trade_calendar.py → stock_basic.py → 其他脚本。

任务执行模式（通过 DATA_JOB_EXECUTION_MODE 环境变量控制）：

inline（默认开发模式）：在 Web 进程中直接执行，无需 Celery
celery（默认生产模式）：通过 Celery Worker 异步执行，需要 Redis

2.8 Text2SQL 自然语言查询模块

API 蓝图前缀：/api

核心服务：

Text2SQLEngine（services/text2sql_engine.py）：LLM 驱动的自然语言转 SQL 查询
LLMService（services/llm_service.py）：LLM 服务封装（Ollama / OpenAI）

功能说明：

用户可以使用自然语言描述查询需求，系统通过 LLM 将自然语言转换为 SQL 查询语句，直接查询股票数据。

LLM 配置：

配置	默认值	说明
默认提供商	Ollama	本地 LLM 服务
默认模型	qwen2.5-coder	阿里通义千问代码模型
可选提供商	OpenAI API	通过 `LLM_CONFIG` 配置切换

2.9 实时分析报告模块

API 蓝图前缀：/api/realtime-analysis/*

核心服务：

RealtimeReportGenerator（services/realtime_report_generator.py）：分析报告生成器

功能：基于实时行情数据和分析结果，自动生成可视化分析报告，涵盖市场概况、板块表现、个股分析等内容。

2.10 WebSocket 推送模块

API 蓝图前缀：/api/websocket/*

核心服务：

WebSocketPushService（app/websocket/websocket_push_service.py）：WebSocket 推送服务

功能：通过 WebSocket 实现数据的实时推送，支持实时行情、交易信号、风险预警等信息的即时通知。

三、系统架构概览

quantitative_analysis/
├── app/                              # 应用主目录
│   ├── __init__.py                  # Flask 工厂模式 (create_app)
│   ├── extensions.py                # 扩展初始化 (SQLAlchemy, SocketIO, Redis)
│   ├── api/                         # API 蓝图
│   │   ├── analysis_api.py          #   市场分析 API
│   │   ├── ml_factor_api.py         #   ML 因子管理 API
│   │   ├── realtime_analysis.py     #   实时分析 API
│   │   ├── realtime_indicators.py   #   实时指标 API
│   │   ├── realtime_signals.py      #   交易信号 API
│   │   ├── realtime_monitor.py      #   实时监控 API
│   │   ├── realtime_risk.py         #   风险管理 API
│   │   ├── realtime_report.py       #   分析报告 API
│   │   ├── text2sql_api.py          #   自然语言查询 API
│   │   ├── data_jobs_api.py         #   数据下载任务 API
│   │   └── websocket_api.py         #   WebSocket 管理 API
│   ├── models/                      # 数据模型 (SQLAlchemy + Parquet 事件)
│   ├── services/                    # 核心业务服务
│   ├── routes/                      # 页面路由 (HTML 渲染)
│   ├── templates/                   # Jinja2 HTML 模板
│   ├── static/                      # CSS / JS / 图片
│   ├── utils/                       # 数据下载脚本 (通达信/Baostock/Tushare)
│   ├── websocket/                   # WebSocket 推送服务
│   └── services/tongdaxin/          # 通达信行情客户端
├── data/                            # 数据目录
│   ├── stock_minute/                #   分钟级行情 Parquet
│   ├── ml_factor_state/             #   ML 因子状态 Parquet
│   └── realtime_events/            #   实时事件 Parquet
├── tests/                           # 测试用例
├── acceptance/                      # 验收标准与里程碑
├── config.py                        # 配置管理
├── run.py                           # Web 启动入口
└── run_system.py                    # 初始化与诊断工具

四、核心服务一览

服务	文件	职责
FactorEngine	`services/factor_engine.py`	内置因子计算与自定义因子管理
FactorExpressionEngine	`services/factor_expression_engine.py`	白名单校验的自定义因子公式引擎
MLModelManager	`services/ml_models.py`	XGBoost / LightGBM / RandomForest 模型生命周期
ModelTrainingJobService	`services/model_training_job_service.py`	异步训练任务轮询与管理
StockScoringEngine	`services/stock_scoring.py`	因子评分与 ML 评分的综合选股引擎
PortfolioOptimizer	`services/portfolio_optimizer.py`	等权 / 均值方差 / 风险平价 / 因子中性
BacktestEngine	`services/backtest_engine.py`	单策略与多策略回测
Text2SQLEngine	`services/text2sql_engine.py`	LLM 驱动的自然语言转 SQL 查询
RealtimeIndicatorEngine	`services/realtime_indicator_engine.py`	实时技术指标计算
RealtimeTradingSignalEngine	`services/realtime_trading_signal_engine.py`	多策略交易信号生成与融合
RealtimeRiskManager	`services/realtime_risk_manager.py`	风险指标、预警管理、压力测试
DataJobs	`services/data_jobs/`	数据下载任务执行器（Inline / Celery）

五、API 蓝图一览

URL 前缀	功能域
`/api`	主 API — 股票、分析、Text2SQL
`/api/ml-factor/*`	因子管理、ML 模型、选股评分、组合优化
`/api/data-jobs/*`	数据下载任务提交与状态查询
`/api/realtime-analysis/*`	实时指标、信号、监控、风险、报告
`/api/websocket/*`	WebSocket 管理端点

六、快速开始

1. 克隆 & 安装

git clone <repository-url>
cd quantitative_analysis

# 安装依赖
pip install -r requirements.txt

# 如遇兼容问题，使用最小依赖
pip install -r requirements_minimal.txt

2. 下载数据

下载预打包的 A 股历史数据，解压到 data/ 目录即可。也可通过系统内的数据管理页面在线更新。

3. 启动系统

# 启动 Web 服务
python run.py

# 访问地址
http://localhost:5000    # Web 界面
http://localhost:5000/api # API 入口

容器化启动（可选）

cp .env.example .env
docker compose up --build

默认启动 Web + SQLite + Redis；市场数据和 ML 因子状态仍使用本地 Parquet 文件。

七、配置说明

所有配置通过 config.py 管理，支持环境变量 .env 文件覆盖。

配置项	默认值	说明
`DATA_SOURCE`	`parquet`	数据源类型
`SQLALCHEMY_DATABASE_URI`	SQLite	应用状态数据库
`REDIS_HOST / REDIS_PORT`	`localhost:6379`	Redis 连接（Celery/缓存）
`FLASK_ENV`	`development`	运行环境
`DATA_JOB_EXECUTION_MODE`	`inline`	任务执行模式（inline / celery）
`LLM_CONFIG`	Ollama (qwen2.5-coder)	LLM 提供商配置

八、项目定位

✅ 适用场景

量化交易入门学习
策略研究与验证
二次开发与定制
量化系统架构参考

❌ 不适用场景

直接用于实盘交易
高频交易系统

九、版本历史

版本	日期	内容
v1.0.0	2025-06-01	多因子选股系统初始版本：因子管理和计算、机器学习模型集成、组合优化、回测验证引擎、Web 界面和 API 接口
v2.0.0-parquet	2026-06-06	架构升级：从 MySQL 迁移到 Parquet + SQLite，零外部数据库依赖。新增实时行情分析模块、通达信行情接入、日频数据中心、投资组合管理完整闭环、报告可视化、离线数据包

多因子选股系统 — 面向 A 股市场的全链路量化分析平台
MIT License

用hermes agent开发的每日市场简报

Henry Lin — Sat, 23 May 2026 13:59:22 +0000

第一次开发token消耗900K左右，后面复用，只需要调用脚本生成，不需要再消耗token。

📊 每日市场简报

2026-05-21（周四） · 共 5501 只股票

全市场成交额 35,072.80亿

一、涨跌概览

指标	数值
上涨家数	695 (12.6%)
下跌家数	4773
平盘家数	33
涨停数	🔴 40
跌停数	🟢 46
涨跌比	0.15:1

二、成交额 TOP10

排名	代码	名称	成交额(亿)	涨跌幅
1	688008.SH	澜起科技	306.55亿	+1.89%
2	603986.SH	兆易创新	281.99亿	-2.95%
3	300308.SZ	中际旭创	243.95亿	-4.21%
4	600584.SH	长电科技	213.24亿	+0.94%
5	688981.SH	中芯国际	206.02亿	-4.92%
6	688256.SH	寒武纪-U	205.30亿	-5.25%
7	002281.SZ	光迅科技	199.59亿	-8.08%
8	002384.SZ	东山精密	193.85亿	-7.19%
9	000988.SZ	华工科技	193.22亿	-5.79%
10	300502.SZ	新易盛	192.60亿	-3.74%

三、行业涨跌排名

排名	行业	均涨幅	上涨	下跌	总数
1	空运	+1.98%	7	0	8
2	公共交通	+0.93%	4	3	7
3	银行	+0.17%	20	18	42
4	白酒	+0.10%	7	12	19
5	路桥	-0.01%	10	7	19
6	水运	-0.03%	9	10	19
7	证券	-0.14%	14	31	50
8	摩托车	-0.24%	3	8	11
9	玻璃	-0.32%	5	14	19
10	汽车整车	-0.35%	5	16	22
11	生物制药	-0.47%	30	50	80
12	旅游服务	-0.57%	2	5	8
13	港口	-0.58%	4	11	16
14	酒店餐饮	-0.67%	3	6	9
15	机场	-0.78%	0	4	5
16	红黄酒	-0.90%	1	8	9
17	旅游景点	-0.91%	3	15	18
18	汽车配件	-0.91%	81	174	258
19	保险	-0.94%	0	5	5
20	纺织机械	-1.13%	3	8	11
21	火力发电	-1.26%	5	26	31
22	公路	-1.30%	0	3	3
23	铁路	-1.48%	0	7	7
24	水力发电	-1.49%	5	17	22
25	家用电器	-1.51%	24	66	91
26	电器连锁	-1.65%	0	2	2
27	普钢	-1.67%	2	24	26
28	机床制造	-1.69%	5	19	24
29	工程机械	-1.77%	9	29	38
30	化学制药	-1.80%	25	120	147
31	园区开发	-1.81%	1	13	14
32	中成药	-1.88%	5	63	68
33	医疗保健	-1.93%	27	153	181
34	水泥	-2.06%	4	19	23
35	多元金融	-2.11%	0	23	23
36	机械基件	-2.23%	34	100	134
37	服饰	-2.26%	10	54	64
38	特种钢	-2.26%	0	10	10
39	造纸	-2.29%	6	28	34
40	仓储物流	-2.31%	2	43	46
41	日用化工	-2.33%	2	18	20
42	汽车服务	-2.33%	1	9	10
43	农业综合	-2.34%	2	40	42
44	新型电力	-2.35%	5	25	30
45	食品	-2.43%	5	87	93
46	医药商业	-2.44%	0	37	37
47	百货	-2.44%	3	32	35
48	水务	-2.45%	0	14	15
49	家居用品	-2.49%	8	63	71
50	饲料	-2.54%	1	22	23
51	元器件	-2.56%	58	242	300
52	纺织	-2.58%	6	36	42
53	石油加工	-2.59%	1	13	14
54	渔业	-2.61%	0	7	7
55	小金属	-2.67%	8	56	64
56	铝	-2.70%	4	26	30
57	啤酒	-2.71%	0	8	8
58	轻工机械	-2.74%	0	7	7
59	黄金	-2.86%	0	10	10
60	农药化肥	-2.91%	3	55	58
61	影视音像	-2.92%	0	35	35
62	乳制品	-2.95%	0	21	21
63	煤炭开采	-2.97%	1	25	26
64	农用机械	-3.00%	0	13	13
65	超市连锁	-3.02%	0	8	8
66	林业	-3.06%	0	4	4
67	出版业	-3.07%	0	29	29
68	建筑工程	-3.12%	7	121	128
69	电器仪表	-3.14%	14	92	106
70	航空	-3.16%	3	49	52
71	电信运营	-3.17%	0	5	5
72	文教休闲	-3.23%	1	50	51
73	矿物制品	-3.24%	4	30	35
74	环境保护	-3.26%	12	108	120
75	其他商业	-3.26%	1	11	12
76	供气供热	-3.31%	3	42	45
77	塑料	-3.36%	7	71	78
78	种植业	-3.43%	0	19	19
79	广告包装	-3.44%	5	49	54
80	运输设备	-3.47%	1	47	48
81	化纤	-3.47%	1	33	34
82	互联网	-3.52%	4	76	80
83	专用机械	-3.53%	24	254	280
84	商贸代理	-3.54%	1	14	16
85	船舶	-3.55%	0	11	11
86	铜	-3.56%	1	17	18
87	铅锌	-3.59%	1	13	14
88	IT设备	-3.67%	4	73	78
89	电气设备	-3.69%	38	303	342
90	批发业	-3.71%	0	5	5
91	软件服务	-3.77%	23	256	279
92	全国地产	-3.80%	1	25	26
93	染料涂料	-3.84%	4	30	34
94	钢加工	-3.84%	0	23	23
95	其他建材	-3.84%	1	29	30
96	陶瓷	-3.93%	0	5	5
97	房产服务	-3.97%	0	13	13
98	化工原料	-4.00%	16	238	255
99	化工机械	-4.08%	1	12	13
100	焦炭加工	-4.13%	0	7	7
101	综合类	-4.14%	0	19	19
102	橡胶	-4.19%	0	17	17
103	装修装饰	-4.20%	1	24	25
104	商品城	-4.28%	0	4	4
105	区域地产	-4.38%	2	42	44
106	通信设备	-4.69%	11	124	135
107	石油贸易	-4.76%	0	4	4
108	软饮料	-4.83%	0	9	9
109	石油开采	-5.36%	0	21	21
110	半导体	-5.76%	17	176	193

四、主力资金动向

🔥 净流入 TOP5

排名	代码	名称	净流入(亿)
1	002050.SZ	三花智控	3.00亿
2	688008.SH	澜起科技	1.99亿
3	300274.SZ	阳光电源	1.91亿
4	600745.SH	闻泰科技	1.35亿
5	300059.SZ	东方财富	1.03亿

💧 净流出 TOP5

排名	代码	名称	净流出(亿)
1	000725.SZ	京东方Ａ	-4.86亿
2	603986.SH	兆易创新	-3.08亿
3	600522.SH	中天科技	-2.57亿
4	601138.SH	工业富联	-2.18亿
5	002384.SZ	东山精密	-1.87亿

五、特殊形态统计

形态	数量
首板（首次涨停）	32
连板（连续涨停）	8
阳包阴	60
连涨≥3天	142

🎯 连板个股明细

代码	名称	连板天数
603779.SH	威龙股份	7连板
603316.SH	诚邦股份	4连板
000518.SZ	*ST四环	2连板
002442.SZ	龙星科技	2连板
003004.SZ	*ST声迅	2连板
300069.SZ	金利华电	2连板
301139.SZ	元道通信	2连板
600611.SH	大众交通	2连板

🌅 阳包阴个股（TOP20）

代码	名称	涨跌幅
920100.BJ	三协电机	+14.18%
301488.SZ	豪恩汽电	+10.64%
600936.SH	北投科技	+10.08%
600156.SH	华升股份	+10.04%
600360.SH	*ST华微	+10.03%
002888.SZ	惠威科技	+10.00%
002940.SZ	昂利康	+10.00%
603373.SH	安邦护卫	+9.99%
600578.SH	京能电力	+9.95%
000404.SZ	长虹华意	+8.46%
001331.SZ	胜通能源	+7.86%
603037.SH	凯众股份	+6.82%
300276.SZ	三丰智能	+6.60%
600481.SH	双良节能	+6.56%
603040.SH	新坐标	+6.31%
603693.SH	江苏新能	+6.29%
001270.SZ	*ST铖昌	+6.19%
601127.SH	赛力斯	+6.14%
002471.SZ	中超控股	+5.71%
002284.SZ	亚太股份	+5.65%

📈 连涨≥3天个股（TOP10）

代码	名称	3日累计涨幅
301020.SZ	密封科技	+32.53%
603316.SH	诚邦股份	+30.02%
603779.SH	威龙股份	+29.99%
001259.SZ	利仁科技	+29.28%
605288.SH	凯迪股份	+28.65%
688585.SH	上纬新材	+28.29%
688610.SH	埃科光电	+27.44%
300152.SZ	ST新动力	+25.67%
301322.SZ	绿通科技	+25.19%
300715.SZ	凯伦股份	+25.10%

由 Hermes Quant 每日市场简报生成器自动生成

生成时间: 2026-05-23 21:39:23

第 1 章：什么是 FreqAI / 安装与首次运行

Henry Lin — Sun, 17 May 2026 07:21:23 +0000

第 1 章：什么是 FreqAI / 安装与首次运行

对应官方文档：https://www.freqtrade.io/en/stable/freqai/

学习目标

用一句话说清 FreqAI 解决什么问题
理解 FreqAI 的"训练—预测—再训练"循环
完成 FreqAI 依赖安装
跑通一个最小可运行的 LightGBM 回归示例，看到 backtest 报告

1.1 FreqAI 是什么

官方原话：

FreqAI is a software designed to automate a variety of tasks associated with training a predictive machine learning model to generate market forecasts given a set of input features.

人话版：FreqAI 是给 Freqtrade 策略加上"机器学习预测能力"的官方框架。你只要在策略里写几个钩子说明"我要用哪些特征、要预测什么"，FreqAI 就自动负责：

拉数据、构造特征、对齐时间戳
切分训练/测试集
训练模型并保存到磁盘
推理出预测列（&- 开头）回填到 dataframe
按时间窗口自动再训练适应市场漂移

你在策略里像用普通 indicator 一样使用预测列就行。

1.2 FreqAI 的工作循环

┌──────────────────────────────────────────────────────────────────┐
│  Freqtrade Pair Dataframe (OHLCV + base indicators)              │
└──────────────────────────────────────────────────────────────────┘
                          │
                          ▼
┌──────────────────────────────────────────────────────────────────┐
│  Strategy hooks (你写的):                                        │
│    feature_engineering_expand_all   ──▶ %- 特征（按period展开） │
│    feature_engineering_expand_basic ──▶ %- 特征（不随period展开）│
│    feature_engineering_standard     ──▶ %- 特征（最终统一处理） │
│    set_freqai_targets               ──▶ &- 标签（必须）          │
└──────────────────────────────────────────────────────────────────┘
                          │
                          ▼
┌──────────────────────────────────────────────────────────────────┐
│  FreqAI 引擎自动完成：                                           │
│    1. 按 train_period_days 切训练窗口                            │
│    2. 标准化 / 异常值检测 / PCA（可选）                          │
│    3. 调用模型 fit() 训练                                        │
│    4. 保存模型到 user_data/models/<identifier>/                  │
│    5. 在 backtest_period_days 范围内做 predict() 回填到 df       │
└──────────────────────────────────────────────────────────────────┘
                          │
                          ▼
┌──────────────────────────────────────────────────────────────────┐
│  策略 populate_entry_trend / populate_exit_trend：               │
│    用 df["&-s_close"] 或 df["do_predict"] 当作普通指标使用       │
└──────────────────────────────────────────────────────────────────┘
                          │
                          ▼
                       下单 / 平仓

关键点：

%- 前缀：模型用作输入的特征
&- 前缀：模型要预测的目标（label）
无前缀：中间计算列，不进入模型
窗口每滚动 backtest_period_days 天就重新训练一次（实盘里改用 live_retrain_hours）

1.3 安装

FreqAI 不在 Freqtrade 默认依赖里，需要单独装：

conda activate freqtrade

# 进入你的 freqtrade 源码或安装目录
pip install -r requirements-freqai.txt

这会装上：lightgbm、catboost、xgboost、scikit-learn、datasieve 等。

Docker 用户改用镜像 freqtradeorg/freqtrade:stable_freqai 即可。

第 7 章的强化学习需要额外的 requirements-freqai-rl.txt（含 stable-baselines3 与 torch，大约 700MB，按需安装）。

1.4 最小可运行示例

我们要做的事情非常简单：预测未来 12 根 5m K 线的平均收盘价相对当前 close 的偏移。本质是一个回归问题，预测出来的值约等于 "未来一小时的预期回报率"。

策略文件：examples/ch01/FreqaiHelloStrategy.py

配置文件：examples/ch01/config_ch01.json

1.4.1 策略代码精读

class FreqaiHelloStrategy(IStrategy):
    timeframe = "5m"
    can_short = False
    process_only_new_candles = True
    startup_candle_count = 40           # >= indicator_periods_candles 最大值 * 2

    def feature_engineering_expand_all(self, dataframe, period, metadata, **kwargs):
        # 每个 period（10, 20）都会被自动展开一遍
        dataframe[f"%-rsi-period_{period}"] = ta.RSI(dataframe, timeperiod=period)
        dataframe[f"%-ema-period_{period}"] = ta.EMA(dataframe, timeperiod=period)
        return dataframe

    def set_freqai_targets(self, dataframe, metadata, **kwargs):
        # 必须实现：定义 &- 开头的列作为预测目标
        horizon = self.freqai_info["feature_parameters"]["label_period_candles"]
        dataframe["&-s_close"] = (
            dataframe["close"].shift(-horizon).rolling(horizon).mean()
            / dataframe["close"] - 1
        )
        return dataframe

    def populate_indicators(self, dataframe, metadata):
        # 把 freqai 引擎接进来——这是必须的一行
        dataframe = self.freqai.start(dataframe, metadata, self)
        return dataframe

    def populate_entry_trend(self, dataframe, metadata):
        # 预测值 > 0 且预测置信度 ok 就开多
        dataframe.loc[
            (dataframe["&-s_close"] > 0.002) & (dataframe["do_predict"] == 1),
            "enter_long",
        ] = 1
        return dataframe

    def populate_exit_trend(self, dataframe, metadata):
        dataframe.loc[dataframe["&-s_close"] < 0, "exit_long"] = 1
        return dataframe

记住两个魔法字段：

do_predict：FreqAI 自动加在 dataframe 上的"这一行预测是否有效"标记（1 表示在 DI 阈值内可信）
&-s_close：你定义的 label 在推理阶段会被同名列自动覆盖为预测值

1.4.2 配置文件精读

config_ch01.json 的 freqai 块：

"freqai": {
    "enabled": true,
    "identifier": "ch01_hello",
    "train_period_days": 15,
    "backtest_period_days": 5,
    "live_retrain_hours": 0,
    "purge_old_models": 2,
    "feature_parameters": {
        "include_timeframes": ["5m"],
        "include_corr_pairlist": [],
        "label_period_candles": 12,
        "include_shifted_candles": 2,
        "indicator_periods_candles": [10, 20]
    },
    "data_split_parameters": { "test_size": 0.2, "shuffle": false },
    "model_training_parameters": { "n_estimators": 300 }
}

最小可运行配置：单时间框架、无相关币种、两个指标周期。

1.5 运行命令

conda activate freqtrade

# 1. 预下载数据（带 35 天 buffer，给 train_period_days=15 + startup=40 留余地）
freqtrade download-data \
  --exchange binance \
  --pairs BTC/USDT ETH/USDT \
  --timeframes 5m \
  --timerange 20240501-20240801

# 2. 跑回测
freqtrade backtesting \
  --config freqai-tutorial/examples/ch01/config_ch01.json \
  --strategy FreqaiHelloStrategy \
  --strategy-path freqai-tutorial/examples/ch01 \
  --freqaimodel LightGBMRegressor \
  --timerange 20240601-20240731

期待输出（节选）：

=========== FREQAI ENABLED ===========
Found XX historical candles for BTC/USDT 5m
Training model for BTC/USDT
...
============================ BACKTESTING REPORT ============================
| Pair        | Entries | Avg Profit % | ... |
| BTC/USDT    | XX      | X.XX         | ... |

只要看到 "Training model for ..." 与最终的 BACKTESTING REPORT 表格，就说明跑通了。

1.6 常见错误

报错	原因	解决
`KeyError: 'freqai'` 或 `freqai_info is None`	配置里没启用 freqai 块	检查 `freqai.enabled: true`，且 `freqaimodel` 命令行参数已传
`No model found in user_data/models/<identifier>/...`	训练窗口不够 / 数据没下载完	把 `--timerange` 起点往后挪到 `train_period_days + startup` 之后
`KeyError: '&-s_close'` 在 populate_entry_trend	忘了在 populate_indicators 里调 `self.freqai.start(...)`	加上那一行
`Cannot use VolumePairList with FreqAI`	用了动态币池	改成静态 `StaticPairList` 或 `ShufflePairList`

1.7 验证情况

✅ 本章示例在 conda activate freqtrade 环境下完整跑通了 backtesting（见教程仓库 README）。如果你的环境跑不起来，多半是数据没下全或 freqtrade 版本太老（建议 ≥ 2024.x）。

1.8 小结与下一章

你已经会跑一个最小 FreqAI 策略了。但很多配置字段是直接照抄的，第 2 章会逐项拆解 freqai 块每个字段的含义，让你能根据自己的需求改配置。

为什么长期做量化，一定要先搭建自己的“数据中台”

Henry Lin — Tue, 12 May 2026 11:24:30 +0000

为什么长期做量化，一定要先搭建自己的数据中台

很多人做量化，第一步就开始研究策略。

今天学一个 MACD，明天学一个 AI 选股，后天开始回测。结果半年后，策略越来越多，代码越来越乱，数据越来越碎，整个系统根本没法扩展。

尤其是个人开发者、独立量化团队，经常会踩一个坑：

把量化系统当成了策略工程，而不是数据工程。

但说实话，真正长期跑得起来的量化系统，重心其实不在策略上，而在数据层。

为什么数据底座这么重要？

量化系统本质上就是数据驱动。所有策略、因子、回测、AI 模型、风控、选股，全都建在数据之上。

数据层一旦出问题，后面什么都跟着乱：

指标没法统一，同一个 PE 不同来源数值不一样
股票代码混乱，有的表叫 code，有的叫 symbol，有的叫 ts_code
时间周期对不上，日线周线分钟线混在一起
财务数据缺失，财报更新不及时
因子重复计算，好几个策略各算一遍一样的东西
回测和实盘用的数据不一致，结果根本没法复现

很多人刚开始只有几十只股票的数据，觉得 CSV 就够了。但做半年以后你会发现，日线、分钟线、基本面、财报、资金流、技术指标、因子库、AI 特征、新闻舆情、宏观数据……数据量增长得比你想的快得多。这时候如果没有统一的数据管理方式，后面几乎没法维护。

一个成熟量化系统需要哪些数据？

长期来看，建议至少把下面这些数据管起来：

                    ┌─────────────────────┐
                    │    数据中台 / DataHub │
                    └─────────────────────┘
                               │
 ┌────────────────────────────────────────────────────────┐
 │                                                        │
 │ • daily_history   (日线行情 OHLCV)                      │
 │ • daily_basic    (基本面 PE/PB/换手率)                  │
 │ • moneyflow      (资金流向)                             │
 │ • stk_factor     (技术指标 MACD/KDJ/RSI)                │
 │ • stock_basic    (股票基础信息)                         │
 │ • balance_sheet  (资产负债表)                           │
 │ • income_statement (利润表)                             │
 │ • cash_flow      (现金流量表)                           │
 │                                                        │
 └────────────────────────────────────────────────────────┘
                               │
        ┌──────────────────────────────────┐
        │                                  │
        │  因子系统 / 回测 / AI / 策略引擎   │
        │                                  │
        └──────────────────────────────────┘

为什么建议一开始就搭？

有开发者会说："我策略还没跑通，先不用那么复杂吧。"

这想法挺正常的，但后期再重构数据层，代价真的很大。

量化有个特点：数据只会越来越多。你今天可能只有 500 只股票的日线，但早晚要加全市场、分钟线、Tick、期货、ETF、美股、港股、AI 特征、另类数据。

如果一开始没设计好数据结构，后面大概率会遇到：表结构崩坏、查询变慢、指标重复计算、Redis 缓存失控、CSV 满天飞、Python 脚本互相依赖改一个挂一片。等你想整理的时候，已经理不清了。

量化的核心其实不是策略，是数据复用

大部分刚入门的人觉得策略最重要。但实际上，成熟的量化团队花在数据一致性、数据清洗、数据标准化、因子复用、特征工程上的精力，远比调参多。

原因很简单：一份数据可以反复用。

比如同一份 daily_history，动量策略用得着，AI 训练用得着，因子分析用得着，行业轮动用得着，回测用得着，择时用得着，可视化展示用得着，选股器也用得着。

所以长期来看，真正有价值的是持续积累的数据资产，不是某一个短期能赚钱的策略。策略会过时，数据不会。

AI 和量化的关系

AI 深度介入量化已经是大趋势了。但 AI 有个硬需求：它要吃数据，而且是干净、结构化、能对齐的数据。

没有稳定的数据底座，特征工程做不了，时序训练做不了，多周期融合做不了，Walk-Forward 做不了，因子挖掘做不了，AutoML 也做不了。

我见过不少人，最后卡住的不是策略能力，是数据根本接不起来。

个人开发者推荐的技术架构

个人或者小团队没必要上来就搞 Hadoop。以下这些够用了：

数据存储： PostgreSQL / ClickHouse / DuckDB / Parquet

缓存层： Redis

任务调度： Airflow / Celery / Cron

数据处理： Python + Pandas + Polars

因子计算： TA-Lib / vectorbt / Qlib

两条不同的路

很多人的路径是这样的：

策略 → 回测 → 亏钱 → 放弃

而把数据当核心资产的人，路径大概是这样：

数据底座 → 因子平台 → 回测框架 → AI 特征工程 → 自动选股 → 多策略组合 → 风控系统 → 实盘交易

区别就在于你把数据当成附属品，还是当成核心资产。

总结

长期做量化，与其急着写策略、做回测、研究 AI，不如先把数据中台搭起来。

策略会失效，市场会变，模型会过时。但数据资产会一直积累，越往后越值钱。

量化这件事，拼到最后，壁垒往往不是什么策略秘密，而是你手里有没有一套长期积累、持续迭代的数据底座。