从零搭建量化分析系统之热点追踪-01-技术方案
版本：v0.1（MVP）
日期：2026-07-13
目标周期：4 周 MVP → 2 周优化 → 持续迭代
一、项目概述
1.1 系统定位
从通达信盘口/逐笔数据中，实时捕获市场热点股票及板块资金异动，输出量化的"热点指数"（HotScore），辅助交易决策。
1.2 核心目标
| 层级 | 目标 | 说明 |
|---|---|---|
| 近期 | 未来 5–30 分钟是否会成为市场热点 | 可验证、有交易价值 |
| 中期 | 板块联动与龙头识别 | 提供板块维度的资金流向视图 |
| 远期 | AI 预测热度提升 | 在规则引擎基础上叠加模型 |
不做：预测涨停、预测未来一周股价。
最重要的原则：Evaluation 比 AI 更重要。
整个系统的价值取决于一个问题：HotScore 选出来的股票，到底准不准？
评估层是这个问题的唯一裁判。它从第 2 周就开始运行，并且永不停止。
AI 模型只有在评估达标后才有资格启动。
1.3 整体架构
┌──────────────────────────────────────────────────────────────────┐
│ 前端展示层 │
│ 实时热点排行 │ 板块热力图 │ 个股资金面板 │ 异动告警 │ 评估看板 │
└──────────────────────────┬───────────────────────────────────────┘
│ WebSocket / SSE
┌──────────────────────────┴───────────────────────────────────────┐
│ API 服务层 (FastAPI) │
│ /hotspot/* /sector/* /alert/* /eval/* /model/* │
└──────────────────────────┬───────────────────────────────────────┘
│
┌──────────────────────────┴───────────────────────────────────────┐
│ 计算引擎层 │
│ ┌────────────┐ ┌────────────┐ ┌────────────┐ ┌──────────────┐ │
│ │ FeatureEng │ │ HotScore │ │ SectorEng │ │ AI Predictor │ │
│ │ 特征计算 │ │ 热点评分 │ │ 板块联动 │ │ 模型推理 │ │
│ └────────────┘ └────────────┘ └────────────┘ └──────────────┘ │
└──────────────────────────┬───────────────────────────────────────┘
│
┌──────────────────────────┴───────────────────────────────────────┐
│ 数据管道层 │
│ ┌──────────────┐ ┌──────────────┐ ┌────────────────────────┐ │
│ │ TickCollector│ │ OrderBookCol │ │ SectorCollector │ │
│ │ 逐笔成交采集 │ │ 五档盘口采集 │ │ 板块/涨停/情绪数据采集 │ │
│ └──────┬───────┘ └──────┬───────┘ └────────────┬───────────┘ │
│ └────────────────┴──────────────────────┘ │
│ DataBus (内存队列 / Redis Stream) │
└──────────────────────────┬───────────────────────────────────────┘
│
┌──────────────────────────┴───────────────────────────────────────┐
│ 存储层 │
│ Redis (实时状态/排行) │ DuckDB (历史特征/回测) │ MySQL (配置/映射) │
└─────────────────────────────────────────────────────────────────┘
1.4 项目目录结构
market_hotspot/
├── config/
│ ├── settings.yaml # 全局配置
│ ├── sectors.yaml # 板块-个股映射
│ └── hotscore_weights.yaml # HotScore 权重配置（可热更新）
├── collector/ # 数据采集层
│ ├── __init__.py
│ ├── base.py # 采集器基类
│ ├── tick_collector.py # 逐笔成交采集
│ ├── orderbook_collector.py # 五档盘口采集
│ ├── sector_collector.py # 板块行情采集
│ ├── limitup_collector.py # 涨停/炸板数据采集
│ └── manager.py # 采集调度管理器
├── engine/ # 计算引擎层
│ ├── __init__.py
│ ├── feature_engine.py # 特征计算引擎
│ ├── hotscore_engine.py # HotScore 评分引擎
│ ├── sector_engine.py # 板块联动引擎
│ └── alert_engine.py # 异动告警引擎
├── model/ # AI 预测层
│ ├── __init__.py
│ ├── label_builder.py # 标签构建（严防未来函数）
│ ├── trainer.py # 离线训练
│ ├── predictor.py # 在线推理
│ └── saved/ # 模型文件目录
├── evaluator/ # 评估层（核心模块）
│ ├── __init__.py
│ ├── eval_engine.py # 评估引擎（快照 + 回填 + 汇总）
│ ├── precision.py # 多阈值 Precision / Recall 计算
│ ├── baseline.py # 基线对比（随机/成交额Top/涨幅Top）
│ ├── session_analyzer.py # 分时段分析
│ ├── ic_analyzer.py # 特征分量 IC 分析
│ ├── weight_optimizer.py # 基于 IC 的权重自动调优
│ ├── reporter.py # 每日评估报告生成
│ └── eval_storage.py # 评估数据读写（DuckDB）
├── api/ # API 层
│ ├── __init__.py
│ ├── routes_hotspot.py # 热点排行相关路由
│ ├── routes_sector.py # 板块联动相关路由
│ ├── routes_alert.py # 告警相关路由
│ ├── routes_eval.py # 评估数据路由
│ └── ws.py # WebSocket 管理
├── storage/ # 存储层
│ ├── __init__.py
│ ├── redis_store.py # Redis 读写
│ ├── duckdb_store.py # DuckDB 历史数据
│ └── cache.py # 内存缓存（LRU / 时序环形缓冲）
├── common/ # 公共工具
│ ├── __init__.py
│ ├── trading_calendar.py # 交易日历
│ ├── symbol_utils.py # 代码转换（SH600519 ↔ 600519 等）
│ └── logger.py # 统一日志
├── frontend/
│ └── hotspot_dashboard.html # 实时热点监控面板
├── tests/
│ ├── test_feature_engine.py
│ ├── test_hotscore.py
│ └── test_collector.py
├── docs/
│ ├── 热点侦测系统.pdf # 原始规划书
│ └── 技术方案.md # 本文档
├── main.py # 启动入口
└── requirements.txt
1.5 技术栈总览
| 类别 | 选型 | 版本要求 | 说明 |
|---|---|---|---|
| 语言 | Python | ≥3.10 | asyncio 原生支持 |
| Web 框架 | FastAPI | ≥0.100 | 异步、WebSocket 原生支持 |
| 行情数据 | pytdx | ≥1.72 | 逐笔成交 + 盘口（通达信协议） |
| 板块/涨停 | akshare | ≥1.12 | 东方财富数据源 |
| 实时存储 | Redis | ≥7.0 | Sorted Set + Stream |
| 历史存储 | DuckDB | ≥0.10 | 列式分析，复用现有基础设施 |
| 关系存储 | MySQL | ≥8.0 | 配置、映射表，复用现有库 |
| 定时调度 | APScheduler | ≥3.10 | 多频率任务调度 |
| 数据处理 | pandas + numpy | pandas≥2.0 | 滚动窗口特征计算 |
| ML 模型 | LightGBM | ≥4.0 | 盘中实时推理，速度优于 XGBoost |
| 前端 | 原生 HTML/JS | — | 与 stock_pool_app 风格统一 |
二、数据采集层详细设计
2.1 采集器基类
所有采集器继承统一基类，确保一致的生命周期管理、错误处理和监控。
class BaseCollector(ABC):
"""采集器基类"""
def __init__(self, name: str, interval_seconds: float):
self.name = name
self.interval_seconds = interval_seconds
self.is_running = False
self.stats = CollectorStats() # 采集统计
@abstractmethod
async def connect(self) -> None: ...
@abstractmethod
async def disconnect(self) -> None: ...
@abstractmethod
async def fetch_once(self) -> list[dict]: ...
async def run_loop(self):
"""主循环：fetch → publish → sleep"""
while self.is_running:
try:
t0 = time.monotonic()
records = await self.fetch_once()
await self.publish(records)
self.stats.record_success(len(records), time.monotonic() - t0)
except Exception as e:
self.stats.record_error(e)
await self.handle_error(e)
await asyncio.sleep(self.interval_seconds)
async def publish(self, records: list[dict]):
"""发布到 DataBus（内存队列 + 可选 Redis Stream）"""
await DataBus.publish(self.name, records)
2.2 逐笔成交采集器 (TickCollector)
数据源：pytdx 的 get_transaction_data 接口。
采集策略：
初始化阶段（开盘前）:
1. 从 Redis/配置 加载今日关注列表（默认：昨日成交额 Top300 + 股票池标的）
2. 建立 pytdx 连接池（3~5 个连接）
盘中循环（每 3 秒）:
1. 将关注列表分片，分配给连接池中各连接并发拉取
2. 对每只股票记录上次拉取的最后一条索引，增量拉取新成交
3. 原始逐笔 → 解析方向（B/S）→ 发布到 DataBus
盘中动态扩展（每 1 分钟）:
1. 检查是否有新股票触发"粗筛条件"（量比>2 或涨幅>3%）
2. 将新标的加入关注列表
逐笔数据结构：
@dataclass
class TickRecord:
symbol: str # 股票代码 (如 "600519")
market: int # 市场 (0=深圳, 1=上海)
timestamp: datetime # 成交时间
price: float # 成交价
volume: int # 成交量（手）
amount: float # 成交额（元）
direction: int # 方向：1=主动买, -1=主动卖, 0=中性
order_kind: str # 订单类型
连接池管理：
class TdxConnectionPool:
"""pytdx 连接池，支持自动重连和服务器切换"""
def __init__(self, pool_size: int = 5):
self.pool_size = pool_size
self.connections: list[TdxHq_API] = []
self.server_list = [
("119.147.212.81", 7709),
("14.17.75.71", 7709),
("14.215.128.18", 7709),
("59.173.18.140", 7709),
("218.75.126.9", 7709),
]
self.current_server_idx = 0
async def get_connection(self) -> TdxHq_API:
"""从池中获取可用连接，不可用时自动重建"""
...
async def execute_batch(
self, symbols: list[str], fetch_fn: Callable
) -> dict[str, Any]:
"""将 symbols 分片，分配给各连接并发执行"""
chunk_size = math.ceil(len(symbols) / self.pool_size)
chunks = [symbols[i:i+chunk_size]
for i in range(0, len(symbols), chunk_size)]
tasks = [
self._fetch_chunk(conn, chunk, fetch_fn)
for conn, chunk in zip(self.connections, chunks)
]
results = await asyncio.gather(*tasks, return_exceptions=True)
return self._merge_results(results)
pytdx 同步转异步的处理：
pytdx 本身是同步阻塞的，需要用 asyncio.to_thread 或 loop.run_in_executor 包装：
async def fetch_once(self) -> list[TickRecord]:
raw_data = await asyncio.to_thread(
self._sync_fetch_all, self.watch_list
)
return self._parse_ticks(raw_data)
def _sync_fetch_all(self, symbols: list[str]) -> dict:
"""同步批量拉取（在线程池中执行）"""
result = {}
for symbol in symbols:
market = 1 if symbol.startswith("6") else 0
data = self.api.get_transaction_data(
market, symbol, start=self.last_index.get(symbol, 0), count=1000
)
if data:
result[symbol] = data
self.last_index[symbol] = (
self.last_index.get(symbol, 0) + len(data)
)
return result
2.3 五档盘口采集器 (OrderBookCollector)
数据源：pytdx 的 get_security_quotes 接口（单次最多可查 80 只）。
采集策略：
每 3 秒:
1. 将关注列表按 80 只一组分批
2. 并发拉取各批次五档行情
3. 解析买1~买5、卖1~卖5 的价格和挂单量
4. 计算快照级衍生字段（买卖盘总量、中间价等）
5. 发布到 DataBus
连续快照差分（每分钟聚合）:
1. 对比相邻两个快照的挂单量变化，估算撤单量
2. 统计买盘增量 vs 卖盘增量
盘口数据结构：
@dataclass
class OrderBookSnapshot:
symbol: str
timestamp: datetime
last_price: float # 最新价
bid_prices: list[float] # 买1~买5价格
bid_volumes: list[int] # 买1~买5挂单量（手）
ask_prices: list[float] # 卖1~卖5价格
ask_volumes: list[int] # 卖1~卖5挂单量（手）
total_volume: int # 当日总成交量
total_amount: float # 当日总成交额
open_price: float
high_price: float
low_price: float
pre_close: float
@property
def bid_total(self) -> int:
return sum(self.bid_volumes)
@property
def ask_total(self) -> int:
return sum(self.ask_volumes)
@property
def bid_ask_imbalance(self) -> float:
"""买卖盘失衡度：>1 买方强势，<1 卖方强势"""
if self.ask_total == 0:
return float('inf')
return self.bid_total / self.ask_total
2.4 板块行情采集器 (SectorCollector)
数据源：akshare 的 stock_board_concept_name_em（概念板块）和 stock_board_industry_name_em（行业板块）。
采集策略：
盘前（09:15）:
1. 拉取全量板块列表，更新板块-个股映射缓存
每 30 秒:
1. 拉取概念板块实时涨幅排行
2. 拉取行业板块实时涨幅排行
3. 发布板块级聚合数据到 DataBus
每 1 分钟:
1. 拉取涨停数据（东方财富涨停板接口）
2. 统计各板块涨停家数、炸板家数
板块数据结构：
@dataclass
class SectorSnapshot:
sector_name: str # 板块名称
sector_code: str # 板块代码
sector_type: str # "concept" | "industry"
change_pct: float # 涨跌幅 %
total_amount: float # 板块总成交额（亿）
up_count: int # 上涨家数
down_count: int # 下跌家数
limit_up_count: int # 涨停家数
limit_down_count: int # 跌停家数
leader_symbol: str # 领涨股代码
leader_change_pct: float # 领涨股涨幅
timestamp: datetime
2.5 涨停/炸板采集器 (LimitUpCollector)
数据源：akshare 的 stock_zt_pool_em（涨停池）、stock_zt_pool_zbgc_em（炸板池）。
@dataclass
class LimitUpRecord:
symbol: str
name: str
limit_up_time: str # 首次封板时间 "09:35:21"
last_seal_time: str # 最新封板时间
open_count: int # 打开次数（炸板次数）
seal_amount: float # 封单金额（万）
seal_ratio: float # 封单占流通市值比
continuous_days: int # 连板天数
sectors: list[str] # 所属概念板块
timestamp: datetime
2.6 采集调度管理器 (CollectorManager)
统一管理所有采集器的生命周期，根据交易时段自动启停。
class CollectorManager:
"""采集调度管理器"""
def __init__(self, config: dict):
self.collectors: dict[str, BaseCollector] = {}
self.config = config
def register(self, collector: BaseCollector):
self.collectors[collector.name] = collector
async def start_all(self):
"""根据交易日历启动采集"""
if not is_trading_day(date.today()):
logger.info("非交易日，跳过采集")
return
for collector in self.collectors.values():
await collector.connect()
# 按各自频率启动采集循环
tasks = [
asyncio.create_task(c.run_loop())
for c in self.collectors.values()
]
# 等待交易时段结束
await self._wait_until_market_close()
for c in self.collectors.values():
c.is_running = False
await asyncio.gather(*tasks)
# 收盘后持久化当日数据
await self._persist_daily_data()
async def _wait_until_market_close(self):
"""等待至 15:05（留 5 分钟缓冲收尾数据）"""
...
交易时段控制：
09:15 - 09:25 集合竞价阶段：仅采集盘口，不采集逐笔
09:25 - 09:30 等待开盘
09:30 - 11:30 上午连续竞价：全采集器启动
11:30 - 13:00 午休：暂停采集
13:00 - 15:00 下午连续竞价：全采集器启动
15:00 - 15:05 收盘缓冲：完成最后一批数据落盘
2.7 DataBus（数据总线）
内存中的发布-订阅总线，解耦采集器与计算引擎。
class DataBus:
"""进程内数据总线，基于 asyncio.Queue"""
_channels: dict[str, list[asyncio.Queue]] = {}
@classmethod
def subscribe(cls, channel: str) -> asyncio.Queue:
queue = asyncio.Queue(maxsize=10000)
cls._channels.setdefault(channel, []).append(queue)
return queue
@classmethod
async def publish(cls, channel: str, records: list[dict]):
for queue in cls._channels.get(channel, []):
try:
queue.put_nowait(records)
except asyncio.QueueFull:
# 消费者太慢，丢弃最旧的数据
queue.get_nowait()
queue.put_nowait(records)
@classmethod
async def publish_to_redis(cls, channel: str, records: list[dict]):
"""可选：同时写入 Redis Stream 做持久化和跨进程共享"""
...
频道定义：
| 频道名 | 生产者 | 数据频率 | 消费者 |
|---|---|---|---|
tick |
TickCollector | ~3秒 | FeatureEngine |
orderbook |
OrderBookCollector | ~3秒 | FeatureEngine |
sector |
SectorCollector | ~30秒 | SectorEngine |
limitup |
LimitUpCollector | ~1分钟 | SectorEngine, AlertEngine |
2.8 采集监控指标
每个采集器实时维护以下统计：
@dataclass
class CollectorStats:
total_fetches: int = 0 # 总采集次数
total_records: int = 0 # 总记录数
error_count: int = 0 # 错误次数
last_fetch_time: float = 0 # 上次采集耗时（秒）
avg_fetch_time: float = 0 # 平均采集耗时
last_error: str = "" # 最近一次错误信息
last_success_at: datetime = None
reconnect_count: int = 0 # 重连次数
通过 GET /api/collector/stats 暴露为监控端点。
三、特征计算引擎详细设计
3.1 整体架构
FeatureEngine 订阅 DataBus 的 tick 和 orderbook 频道，维护每只股票的时序缓冲区，实时计算 4 大类特征。
DataBus (tick/orderbook)
│
▼
┌─────────────────────────────────────────┐
│ FeatureEngine │
│ │
│ ┌──────────────────────────────┐ │
│ │ TimeSeriesBuffer (per stock)│ │
│ │ - tick_buffer (环形, 保留30分钟)│ │
│ │ - ob_buffer (环形, 保留30分钟) │ │
│ │ - minute_bars (当日分钟K) │ │
│ └──────────────┬───────────────┘ │
│ │ │
│ ┌──────────────┴───────────────┐ │
│ │ 特征计算管道 │ │
│ │ 1. 成交特征 (VolumeFeatures) │ │
│ │ 2. 资金主动性 (FlowFeatures) │ │
│ │ 3. 盘口结构 (OrderBookFeats) │ │
│ │ 4. 联动特征 (LinkageFeatures) │ │
│ └──────────────┬───────────────┘ │
│ │ │
│ ┌──────────────┴───────────────┐ │
│ │ FeatureSnapshot (per stock) │ │
│ │ → 发布到 DataBus "features" │ │
│ │ → 写入 Redis (实时查询) │ │
│ └──────────────────────────────┘ │
└─────────────────────────────────────────┘
3.2 时序缓冲区
为每只股票维护高效的环形缓冲区，避免无限增长：
class TimeSeriesBuffer:
"""单只股票的时序数据缓冲区"""
def __init__(self, symbol: str, max_minutes: int = 30):
self.symbol = symbol
self.max_minutes = max_minutes
# 逐笔成交缓冲（原始粒度，保留最近 max_minutes 分钟）
self.ticks: deque[TickRecord] = deque()
# 盘口快照缓冲（每 3 秒一条）
self.orderbooks: deque[OrderBookSnapshot] = deque()
# 分钟 K 线（当日全部保留）
self.minute_bars: list[MinuteBar] = []
# 预聚合：每分钟统计值（减少重复计算）
self.minute_agg: dict[str, MinuteAggregate] = {}
def append_tick(self, tick: TickRecord):
self.ticks.append(tick)
self._evict_old(self.ticks)
self._update_minute_agg(tick)
def append_orderbook(self, ob: OrderBookSnapshot):
self.orderbooks.append(ob)
self._evict_old(self.orderbooks)
def _evict_old(self, buf: deque):
cutoff = datetime.now() - timedelta(minutes=self.max_minutes)
while buf and buf[0].timestamp < cutoff:
buf.popleft()
def get_recent_ticks(self, minutes: int) -> list[TickRecord]:
cutoff = datetime.now() - timedelta(minutes=minutes)
return [t for t in self.ticks if t.timestamp >= cutoff]
分钟预聚合结构（减少实时计算开销）：
@dataclass
class MinuteAggregate:
"""每分钟预聚合统计"""
minute_key: str # "09:35"
total_volume: int = 0 # 总成交量（手）
total_amount: float = 0 # 总成交额（元）
buy_volume: int = 0 # 主动买成交量
sell_volume: int = 0 # 主动卖成交量
buy_amount: float = 0 # 主动买成交额
sell_amount: float = 0 # 主动卖成交额
trade_count: int = 0 # 成交笔数
large_order_amount: float = 0 # 大单成交额（单笔>50万）
max_price: float = 0
min_price: float = 0
vwap: float = 0 # 成交量加权均价
3.3 特征定义与计算
3.3.1 成交特征 (VolumeFeatures)
| 特征名 | 计算方式 | 窗口 | 说明 |
|---|---|---|---|
volume_surge_5m |
最近5分钟成交额 / 前5分钟成交额 | 10min | 5分钟成交额增速（倍数） |
volume_surge_1m |
最近1分钟成交额 / 前1分钟成交额 | 2min | 1分钟成交额增速 |
volume_ratio |
当前累计成交量 / (过去5日同时段平均成交量) | 当日 | 量比 |
turnover_rate |
当日成交量 / 流通股本 | 当日 | 换手率 % |
large_order_ratio |
大单成交额 / 总成交额 | 5min | 大单占比（阈值：单笔>50万元） |
trade_density |
成交笔数 / 时间窗口分钟数 | 5min | 成交密度 |
class VolumeFeatureCalculator:
LARGE_ORDER_THRESHOLD = 500_000 # 50万元
def calculate(self, buf: TimeSeriesBuffer) -> dict:
aggs = buf.minute_agg
now_minute = self._current_minute()
recent_5m = self._sum_amount(aggs, now_minute, 5)
prev_5m = self._sum_amount(aggs, now_minute - 5, 5)
recent_1m = self._sum_amount(aggs, now_minute, 1)
prev_1m = self._sum_amount(aggs, now_minute - 1, 1)
return {
"volume_surge_5m": recent_5m / max(prev_5m, 1),
"volume_surge_1m": recent_1m / max(prev_1m, 1),
"volume_ratio": self._calc_volume_ratio(buf),
"turnover_rate": self._calc_turnover(buf),
"large_order_ratio": self._calc_large_ratio(buf, minutes=5),
"trade_density": self._calc_density(buf, minutes=5),
}
3.3.2 资金主动性特征 (FlowFeatures)
| 特征名 | 计算方式 | 窗口 | 说明 |
|---|---|---|---|
active_buy_amount |
Σ(主动买成交额) | 5min | 主动买入总额 |
active_sell_amount |
Σ(主动卖成交额) | 5min | 主动卖出总额 |
net_active_buy |
主动买 - 主动卖 | 5min | 净主动买入额 |
active_buy_ratio |
主动买额 / 总成交额 | 5min | 主动买占比 |
consecutive_buy_count |
连续主动买单笔数 | 实时 | 当前连续买单计数 |
buy_momentum |
最近5分钟主动买占比 - 前5分钟主动买占比 | 10min | 买入动量变化 |
class FlowFeatureCalculator:
def calculate(self, buf: TimeSeriesBuffer) -> dict:
recent_ticks = buf.get_recent_ticks(minutes=5)
buy_amount = sum(t.amount for t in recent_ticks if t.direction == 1)
sell_amount = sum(t.amount for t in recent_ticks if t.direction == -1)
total = buy_amount + sell_amount
return {
"active_buy_amount": buy_amount,
"active_sell_amount": sell_amount,
"net_active_buy": buy_amount - sell_amount,
"active_buy_ratio": buy_amount / max(total, 1),
"consecutive_buy_count": self._count_consecutive_buys(buf),
"buy_momentum": self._calc_momentum(buf),
}
def _count_consecutive_buys(self, buf: TimeSeriesBuffer) -> int:
"""从最新一笔往回数，连续主动买的笔数"""
count = 0
for tick in reversed(buf.ticks):
if tick.direction == 1:
count += 1
else:
break
return count
3.3.3 盘口结构特征 (OrderBookFeatures)
| 特征名 | 计算方式 | 说明 |
|---|---|---|
bid_ask_imbalance |
买盘总量 / 卖盘总量 | 买卖盘失衡度 |
seal_amount |
涨停价位的买一挂单金额 | 封单金额（仅涨停时有意义） |
seal_duration |
封单持续时间 | 封单稳定时间（分钟） |
cancel_rate |
(t-1挂单量 - t挂单量 + t成交量) / t-1挂单量 | 撤单率（快照差分估算） |
ask_decay_speed |
卖盘总量变化斜率（最近5个快照） | 卖压衰减速度 |
spread |
(卖一 - 买一) / 买一 | 买卖价差 |
depth_ratio |
买一量 / (买一量 + 卖一量) | 盘口深度比 |
class OrderBookFeatureCalculator:
def calculate(self, buf: TimeSeriesBuffer) -> dict:
if not buf.orderbooks:
return self._empty_features()
latest = buf.orderbooks[-1]
features = {
"bid_ask_imbalance": latest.bid_ask_imbalance,
"spread": self._calc_spread(latest),
"depth_ratio": self._calc_depth_ratio(latest),
}
# 需要连续快照的特征
if len(buf.orderbooks) >= 2:
prev = buf.orderbooks[-2]
features["cancel_rate"] = self._estimate_cancel_rate(prev, latest)
features["ask_decay_speed"] = self._calc_ask_decay(buf.orderbooks)
# 涨停相关
if self._is_at_limit_up(latest):
features["seal_amount"] = latest.bid_volumes[0] * latest.bid_prices[0]
features["seal_duration"] = self._calc_seal_duration(buf.orderbooks)
else:
features["seal_amount"] = 0
features["seal_duration"] = 0
return features
3.3.4 联动特征 (LinkageFeatures)
| 特征名 | 计算方式 | 说明 |
|---|---|---|
sector_change_pct |
所属板块涨跌幅 | 板块涨幅 |
sector_up_ratio |
板块内上涨家数 / 总家数 | 板块上涨比例 |
sector_limitup_count |
板块内涨停家数 | — |
sector_volume_surge |
板块成交额增速 | — |
market_sentiment |
全市场涨跌家数比 × 涨停家数权重 | 市场情绪指数 (0-100) |
sector_rank |
该股所属板块在全市场板块涨幅排名 | 板块强度排名 |
联动特征的计算依赖 SectorEngine 的输出，通过 DataBus 的 sector_features 频道获取。
class LinkageFeatureCalculator:
def __init__(self, sector_mapping: dict[str, list[str]]):
self.symbol_to_sectors = self._build_reverse_map(sector_mapping)
def calculate(self, symbol: str, sector_data: dict) -> dict:
sectors = self.symbol_to_sectors.get(symbol, [])
if not sectors:
return self._empty_features()
# 取该股所属的最强板块
best_sector = max(
sectors,
key=lambda s: sector_data.get(s, {}).get("change_pct", 0)
)
sd = sector_data.get(best_sector, {})
return {
"sector_change_pct": sd.get("change_pct", 0),
"sector_up_ratio": sd.get("up_ratio", 0),
"sector_limitup_count": sd.get("limitup_count", 0),
"sector_volume_surge": sd.get("volume_surge", 1),
"market_sentiment": sector_data.get("_market_sentiment", 50),
"sector_rank": sd.get("rank", 999),
}
3.4 特征快照与输出
每只股票的所有特征汇总为一个 FeatureSnapshot，每分钟更新一次：
@dataclass
class FeatureSnapshot:
symbol: str
timestamp: datetime
# 成交特征
volume_surge_5m: float
volume_surge_1m: float
volume_ratio: float
turnover_rate: float
large_order_ratio: float
trade_density: float
# 资金主动性
active_buy_amount: float
active_sell_amount: float
net_active_buy: float
active_buy_ratio: float
consecutive_buy_count: int
buy_momentum: float
# 盘口结构
bid_ask_imbalance: float
seal_amount: float
seal_duration: float
cancel_rate: float
ask_decay_speed: float
spread: float
depth_ratio: float
# 联动特征
sector_change_pct: float
sector_up_ratio: float
sector_limitup_count: int
sector_volume_surge: float
market_sentiment: float
sector_rank: int
def to_dict(self) -> dict:
return asdict(self)
def to_normalize_vector(self) -> np.ndarray:
"""转为归一化向量，供 HotScore / ML 使用"""
...
3.5 性能预算
| 步骤 | 目标耗时 | 说明 |
|---|---|---|
| 单只股票特征计算 | <2ms | 主要是分钟聚合查表 |
| 全部关注列表 (300只) | <600ms | 可并行 |
| 特征发布到 DataBus | <10ms | 内存操作 |
| 特征写入 Redis | <50ms | pipeline 批量写入 |
| 单轮总耗时 | <1秒 | 满足"每分钟计算一次"的要求 |
四、信号层（HotScore 引擎）详细设计
4.1 HotScore 公式
采用规划书推荐的加权评分公式，所有输入特征先做 0-100 截面归一化：
HotScore = w1 × Norm(成交额增速)
+ w2 × Norm(主动买占比)
+ w3 × Norm(板块联动强度)
+ w4 × Norm(换手率)
+ w5 × Norm(封单质量)
默认权重（可通过 hotscore_weights.yaml 热更新）：
| 分量 | 权重 | 归一化输入 | 说明 |
|---|---|---|---|
| w1 成交额增速 | 0.35 | volume_surge_5m |
放量是热点最直接的信号 |
| w2 主动买占比 | 0.25 | active_buy_ratio |
区分真买还是对倒 |
| w3 板块联动强度 | 0.20 | sector_up_ratio × sector_change_pct |
独涨不如板块共振 |
| w4 换手率 | 0.10 | turnover_rate |
参与度 |
| w5 封单质量 | 0.10 | seal_amount × seal_duration |
涨停股的封单质量 |
4.2 截面归一化
对全市场同时刻的特征值做截面排名百分位，消除绝对值量纲差异：
class CrossSectionalNormalizer:
"""截面归一化：将特征转为 0-100 的百分位排名"""
def normalize(
self, feature_name: str, values: dict[str, float]
) -> dict[str, float]:
"""
values: {symbol: raw_value}
returns: {symbol: percentile_score (0-100)}
"""
sorted_items = sorted(values.items(), key=lambda x: x[1])
n = len(sorted_items)
result = {}
for rank, (symbol, _) in enumerate(sorted_items):
result[symbol] = (rank / max(n - 1, 1)) * 100
return result
4.3 HotScore 引擎核心
class HotScoreEngine:
def __init__(self, weights_path: str = "config/hotscore_weights.yaml"):
self.weights = self._load_weights(weights_path)
self.normalizer = CrossSectionalNormalizer()
self._last_reload_time = 0
def compute_all(
self, features: dict[str, FeatureSnapshot]
) -> dict[str, HotScoreResult]:
"""对全部股票计算 HotScore 并返回排序结果"""
self._maybe_reload_weights()
# 提取各维度原始值
raw = {
"volume_surge": {s: f.volume_surge_5m for s, f in features.items()},
"active_buy": {s: f.active_buy_ratio for s, f in features.items()},
"sector_linkage": {
s: f.sector_up_ratio * f.sector_change_pct
for s, f in features.items()
},
"turnover": {s: f.turnover_rate for s, f in features.items()},
"seal_quality": {
s: f.seal_amount * f.seal_duration
for s, f in features.items()
},
}
# 截面归一化
normed = {
dim: self.normalizer.normalize(dim, vals)
for dim, vals in raw.items()
}
# 加权求和
results = {}
for symbol in features:
score = (
self.weights["volume_surge"] * normed["volume_surge"][symbol]
+ self.weights["active_buy"] * normed["active_buy"][symbol]
+ self.weights["sector_linkage"] * normed["sector_linkage"][symbol]
+ self.weights["turnover"] * normed["turnover"][symbol]
+ self.weights["seal_quality"] * normed["seal_quality"][symbol]
)
results[symbol] = HotScoreResult(
symbol=symbol,
score=round(score, 2),
level=self._classify(score),
components={dim: normed[dim][symbol] for dim in normed},
timestamp=features[symbol].timestamp,
)
return results
@staticmethod
def _classify(score: float) -> str:
if score >= 85:
return "strong_hot" # 强热点 → 重点观察
elif score >= 70:
return "potential_hot" # 潜在热点 → 进入观察池
else:
return "normal" # 普通异动 → 暂不跟踪
HotScore 结果结构：
@dataclass
class HotScoreResult:
symbol: str
score: float # 0-100
level: str # "strong_hot" | "potential_hot" | "normal"
components: dict[str, float] # 各维度得分明细
timestamp: datetime
rank: int = 0 # 在全市场的排名（后续填充）
4.4 实时排行榜
class HotRanking:
"""实时热点排行榜，基于 Redis Sorted Set"""
REDIS_KEY = "hotspot:ranking:{date}"
TOP_N = 50
async def update(self, results: dict[str, HotScoreResult]):
"""更新排行榜"""
pipe = self.redis.pipeline()
key = self.REDIS_KEY.format(date=date.today().isoformat())
# 写入 Sorted Set
for symbol, r in results.items():
pipe.zadd(key, {symbol: r.score})
# 设置过期时间（当日有效）
pipe.expire(key, 86400)
await pipe.execute()
async def get_top(self, n: int = 50) -> list[dict]:
"""获取 Top N 热点股"""
key = self.REDIS_KEY.format(date=date.today().isoformat())
items = await self.redis.zrevrange(key, 0, n - 1, withscores=True)
return [{"symbol": s, "score": score, "rank": i + 1}
for i, (s, score) in enumerate(items)]
4.5 HotScore 分数变化追踪
追踪每只股票的 HotScore 历史，识别"急速升温"模式：
class HotScoreTracker:
"""追踪 HotScore 时序变化"""
def __init__(self):
self.history: dict[str, deque[tuple[datetime, float]]] = {}
def record(self, symbol: str, score: float, ts: datetime):
buf = self.history.setdefault(symbol, deque(maxlen=60))
buf.append((ts, score))
def score_delta(self, symbol: str, minutes: int = 5) -> float:
"""最近 N 分钟的分数变化"""
buf = self.history.get(symbol)
if not buf or len(buf) < 2:
return 0
cutoff = datetime.now() - timedelta(minutes=minutes)
old_scores = [s for t, s in buf if t <= cutoff]
if not old_scores:
return 0
return buf[-1][1] - old_scores[-1]
def is_surging(self, symbol: str, threshold: float = 20) -> bool:
"""5分钟内分数上升超过 threshold，视为"急速升温" """
return self.score_delta(symbol, minutes=5) > threshold
五、板块联动引擎详细设计
5.1 板块-个股映射管理
class SectorMapping:
"""板块-个股映射管理器"""
def __init__(self, config_path: str = "config/sectors.yaml"):
self.concept_map: dict[str, list[str]] = {} # 概念→股票列表
self.industry_map: dict[str, list[str]] = {} # 行业→股票列表
self.symbol_concepts: dict[str, list[str]] = {} # 股票→所属概念
self.symbol_industry: dict[str, str] = {} # 股票→所属行业
async def refresh_from_eastmoney(self):
"""从东方财富刷新板块成分股（每日开盘前执行）"""
# akshare: stock_board_concept_cons_em / stock_board_industry_cons_em
...
def get_sector_members(self, sector_name: str) -> list[str]:
"""获取板块成分股列表"""
...
def get_stock_sectors(self, symbol: str) -> list[str]:
"""获取股票所属的全部板块"""
...
5.2 板块级指标聚合
class SectorEngine:
"""板块联动分析引擎"""
def compute_sector_metrics(
self,
sector_name: str,
member_features: dict[str, FeatureSnapshot],
sector_snapshot: SectorSnapshot,
) -> SectorMetrics:
members = list(member_features.values())
up_count = sum(1 for m in members if m.volume_surge_5m > 1)
total = len(members)
return SectorMetrics(
sector_name=sector_name,
change_pct=sector_snapshot.change_pct,
up_ratio=up_count / max(total, 1),
avg_volume_surge=np.mean([m.volume_surge_5m for m in members]),
limitup_count=sector_snapshot.limit_up_count,
total_net_buy=sum(m.net_active_buy for m in members),
heat_score=self._calc_sector_heat(sector_snapshot, members),
leader=self._identify_leader(member_features),
timestamp=datetime.now(),
)
5.3 龙头股识别
龙头股定义为板块内"最先异动 + 资金最强 + 市场辨识度最高"的个股：
def _identify_leader(
self, member_features: dict[str, FeatureSnapshot]
) -> str | None:
"""识别板块龙头"""
if not member_features:
return None
candidates = []
for symbol, feat in member_features.items():
leader_score = (
0.4 * feat.volume_surge_5m # 放量程度
+ 0.3 * feat.active_buy_ratio # 资金强度
+ 0.2 * (feat.seal_amount > 0) # 是否涨停
+ 0.1 * feat.turnover_rate # 活跃度
)
candidates.append((symbol, leader_score))
candidates.sort(key=lambda x: x[1], reverse=True)
return candidates[0][0]
5.4 板块联动检测
def detect_sector_resonance(
self, sector_metrics: dict[str, SectorMetrics]
) -> list[ResonanceEvent]:
"""检测板块共振事件：多个相关板块同时异动"""
events = []
hot_sectors = [
(name, m) for name, m in sector_metrics.items()
if m.heat_score > 70
]
# 两两检查是否存在成分股重叠
for i, (name_a, m_a) in enumerate(hot_sectors):
for name_b, m_b in hot_sectors[i+1:]:
overlap = self._member_overlap(name_a, name_b)
if overlap > 0.3: # 成分股重叠度超过 30%
events.append(ResonanceEvent(
sectors=[name_a, name_b],
overlap_ratio=overlap,
combined_heat=(m_a.heat_score + m_b.heat_score) / 2,
timestamp=datetime.now(),
))
return events
六、AI 预测层详细设计
6.1 设计原则
- 先规则，再模型：HotScore 规则引擎是基础，AI 层在其上做增量预测
- 预测热度而非涨跌：标签是"未来 HotScore 是否提升"而非"涨停"
- 严防未来函数：标签构建代码中加时间戳断言
6.2 标签设计
class LabelBuilder:
"""构建训练标签，严格避免未来函数"""
# 推荐的 3 种标签
LABEL_CONFIGS = {
"hotscore_surge_10m": {
"desc": "未来10分钟 HotScore 是否 > 85",
"horizon_minutes": 10,
"target_fn": lambda future_scores: max(future_scores) > 85,
},
"top20_15m": {
"desc": "未来15分钟是否进入成交额 Top20",
"horizon_minutes": 15,
"target_fn": lambda future_ranks: min(future_ranks) <= 20,
},
"sector_leader_20m": {
"desc": "未来20分钟是否成为板块龙头",
"horizon_minutes": 20,
"target_fn": lambda future_is_leader: any(future_is_leader),
},
}
def build_label(
self,
feature_time: datetime,
label_config: str,
historical_data: pd.DataFrame,
) -> int:
"""
feature_time: 特征计算时间点 T
返回: 1（正样本）/ 0（负样本）
断言保障：只使用 T 之后的数据构建标签
"""
cfg = self.LABEL_CONFIGS[label_config]
horizon = cfg["horizon_minutes"]
future_start = feature_time + timedelta(minutes=1)
future_end = feature_time + timedelta(minutes=horizon)
# 严格断言：标签数据的时间范围
future_data = historical_data[
(historical_data["timestamp"] >= future_start) &
(historical_data["timestamp"] <= future_end)
]
assert future_data["timestamp"].min() > feature_time, \
f"未来函数泄漏！标签数据时间 {future_data['timestamp'].min()} <= 特征时间 {feature_time}"
if future_data.empty:
return -1 # 数据不足，标记为无效
return int(cfg["target_fn"](future_data["value"].tolist()))
6.3 特征工程（用于模型训练的扩展特征）
在实时特征基础上增加时序衍生特征：
class ModelFeatureBuilder:
"""为 ML 模型构建扩展特征"""
def build(self, snapshot: FeatureSnapshot, history: list[FeatureSnapshot]) -> dict:
base = snapshot.to_dict()
# 时序衍生特征
if len(history) >= 5:
scores_5 = [h.volume_surge_5m for h in history[-5:]]
base["volume_surge_trend_5m"] = np.polyfit(range(5), scores_5, 1)[0]
base["volume_surge_std_5m"] = np.std(scores_5)
buy_ratios = [h.active_buy_ratio for h in history[-5:]]
base["buy_ratio_trend_5m"] = np.polyfit(range(5), buy_ratios, 1)[0]
if len(history) >= 10:
scores_10 = [h.volume_surge_5m for h in history[-10:]]
base["volume_surge_trend_10m"] = np.polyfit(range(10), scores_10, 1)[0]
base["volume_surge_max_10m"] = max(scores_10)
base["volume_surge_min_10m"] = min(scores_10)
# 时段特征
minutes_from_open = self._minutes_from_open(snapshot.timestamp)
base["minutes_from_open"] = minutes_from_open
base["is_morning"] = int(minutes_from_open <= 120)
base["is_last_30min"] = int(minutes_from_open >= 210)
# 当前 HotScore（如果有）
base["current_hotscore"] = getattr(snapshot, "hotscore", 0)
return base
6.4 模型训练
class HotspotTrainer:
"""离线模型训练"""
def __init__(self, label_config: str = "hotscore_surge_10m"):
self.label_config = label_config
self.feature_builder = ModelFeatureBuilder()
self.label_builder = LabelBuilder()
def prepare_dataset(
self, start_date: str, end_date: str
) -> tuple[pd.DataFrame, pd.Series]:
"""从 DuckDB 历史数据构建训练集"""
# 1. 加载历史特征快照
features_df = self.storage.load_features(start_date, end_date)
# 2. 构建标签
labels = []
for _, row in features_df.iterrows():
label = self.label_builder.build_label(
feature_time=row["timestamp"],
label_config=self.label_config,
historical_data=features_df,
)
labels.append(label)
features_df["label"] = labels
# 3. 过滤无效标签
valid = features_df[features_df["label"] >= 0]
X = valid.drop(columns=["label", "timestamp", "symbol"])
y = valid["label"]
return X, y
def train(self, X: pd.DataFrame, y: pd.Series) -> lgb.Booster:
"""训练 LightGBM 模型"""
X_train, X_val, y_train, y_val = train_test_split(
X, y, test_size=0.2, shuffle=False # 时序数据不 shuffle
)
train_data = lgb.Dataset(X_train, label=y_train)
val_data = lgb.Dataset(X_val, label=y_val, reference=train_data)
params = {
"objective": "binary",
"metric": ["auc", "binary_logloss"],
"learning_rate": 0.05,
"num_leaves": 31,
"max_depth": 6,
"min_child_samples": 50,
"feature_fraction": 0.8,
"bagging_fraction": 0.8,
"bagging_freq": 5,
"verbose": -1,
}
model = lgb.train(
params,
train_data,
num_boost_round=500,
valid_sets=[val_data],
callbacks=[lgb.early_stopping(50), lgb.log_evaluation(50)],
)
model.save_model(f"model/saved/{self.label_config}_latest.txt")
return model
6.5 在线推理
class HotspotPredictor:
"""盘中在线推理"""
def __init__(self, model_path: str):
self.model = lgb.Booster(model_file=model_path)
self.feature_builder = ModelFeatureBuilder()
def predict_batch(
self,
current_features: dict[str, FeatureSnapshot],
history: dict[str, list[FeatureSnapshot]],
) -> dict[str, PredictionResult]:
"""批量预测"""
rows = []
symbols = []
for symbol, snap in current_features.items():
hist = history.get(symbol, [])
row = self.feature_builder.build(snap, hist)
rows.append(row)
symbols.append(symbol)
if not rows:
return {}
X = pd.DataFrame(rows)
probs = self.model.predict(X)
results = {}
for symbol, prob in zip(symbols, probs):
results[symbol] = PredictionResult(
symbol=symbol,
probability=round(float(prob), 4),
prediction=int(prob > 0.5),
confidence=round(abs(prob - 0.5) * 2, 4),
timestamp=datetime.now(),
)
return results
@dataclass
class PredictionResult:
symbol: str
probability: float # 0-1，成为热点的概率
prediction: int # 0 或 1
confidence: float # 0-1，置信度
timestamp: datetime
6.6 模型更新策略
每日收盘后（15:30）:
1. 将当日盘中数据持久化到 DuckDB
2. 用最近 N 个交易日数据增量训练模型
3. 在最近 3 个交易日数据上验证 AUC，不低于阈值才替换线上模型
每周末:
1. 用全量历史数据重新训练
2. 生成评估报告（特征重要性、标签分布、时段表现等）
七、告警引擎详细设计
7.1 告警规则
class AlertEngine:
"""异动告警引擎"""
ALERT_RULES = [
{
"name": "hot_surge",
"desc": "HotScore 急速升温",
"condition": lambda ctx: ctx["score_delta_5m"] > 25,
"level": "high",
},
{
"name": "volume_explosion",
"desc": "成交额爆发",
"condition": lambda ctx: ctx["volume_surge_5m"] > 5,
"level": "high",
},
{
"name": "sector_resonance",
"desc": "板块共振启动",
"condition": lambda ctx: ctx["sector_heat"] > 80 and ctx["sector_limitup"] >= 3,
"level": "medium",
},
{
"name": "seal_break",
"desc": "涨停板炸板",
"condition": lambda ctx: ctx["was_sealed"] and not ctx["is_sealed"],
"level": "medium",
},
{
"name": "continuous_buy",
"desc": "连续大单买入",
"condition": lambda ctx: ctx["consecutive_buy_count"] > 20 and ctx["active_buy_ratio"] > 0.75,
"level": "low",
},
]
async def check(
self,
features: dict[str, FeatureSnapshot],
scores: dict[str, HotScoreResult],
tracker: HotScoreTracker,
) -> list[Alert]:
alerts = []
for symbol, feat in features.items():
ctx = self._build_context(symbol, feat, scores, tracker)
for rule in self.ALERT_RULES:
if rule["condition"](ctx):
alerts.append(Alert(
symbol=symbol,
rule_name=rule["name"],
description=rule["desc"],
level=rule["level"],
context=ctx,
timestamp=datetime.now(),
))
return alerts
7.2 告警去重与限流
class AlertThrottler:
"""告警去重：同一股票同一规则 N 分钟内不重复告警"""
def __init__(self, cooldown_minutes: int = 5):
self.cooldown = timedelta(minutes=cooldown_minutes)
self.last_alert: dict[str, datetime] = {} # "symbol:rule" → last_time
def should_fire(self, alert: Alert) -> bool:
key = f"{alert.symbol}:{alert.rule_name}"
last = self.last_alert.get(key)
if last and datetime.now() - last < self.cooldown:
return False
self.last_alert[key] = datetime.now()
return True
八、评估层详细设计（核心模块）
设计哲学：Evaluation 比 AI 更重要。
HotScore 的权重是"拍脑袋"定的，AI 模型再花哨也只是锦上添花。
如果没有一套严格的评估体系，我们永远不知道 HotScore 到底准不准。
评估层是整个系统的审判官——它决定这个系统有没有价值、权重该怎么调、
什么时候可以上 AI、什么时候该推倒重来。
8.1 核心问题：HotScore 到底准不准？
回答这个问题的方式只有一个：用未来的事实说话。
每一分钟 → 记录 HotScore Top 20
│
▼
等 10 分钟
│
▼
回头统计这 20 只：
├─ 上涨了几只？
├─ 平均涨了多少？
├─ 几只涨停了？
├─ 几只进了龙虎榜？
├─ 几只进了成交额 Top 20？
└─ Precision / Recall 各多少？
这就是评估层的核心循环。每分钟做一次快照，N 分钟后用事实回填。
不依赖任何主观判断，只看数字。
8.2 评估数据流
T 时刻 T+N 时刻
┌──────────────────────────┐ ┌──────────────────────────┐
│ HotScore 快照 (Snapshot) │ │ 市场事实 (Outcome) │
│ │ │ │
│ Top20 股票列表 │ ───────→ │ 这 20 只的真实表现 │
│ 每只的 HotScore 分数 │ 等 N 分钟 │ 涨跌 / 涨停 / 龙虎榜 │
│ 每只的特征分量 │ │ 成交额排名 / 板块地位 │
│ 当前市场状态 │ │ │
└──────────────────────────┘ └──────────────────────────┘
│ │
└────────────────┬─────────────────────────┘
▼
┌─────────────────────┐
│ EvalRecord 评估记录 │
│ 快照 + 结果 = 一条 │
│ 完整的评估样本 │
└──────────┬──────────┘
│
┌─────────────┼─────────────┐
▼ ▼ ▼
实时仪表盘 每日报告 权重优化输入
8.3 数据结构定义
8.3.1 快照记录（T 时刻拍摄）
@dataclass
class EvalSnapshot:
"""每分钟的 HotScore 快照，用于未来回填评估"""
snapshot_id: str # "{date}_{HHmm}" 如 "2026-07-14_0935"
timestamp: datetime # 快照时间
market_state: MarketState # 当时市场状态
# Top N 列表（默认 Top20，可配置）
top_symbols: list[str] # 进入 Top N 的股票代码
top_scores: dict[str, float] # {symbol: hotscore}
top_components: dict[str, dict] # {symbol: {维度: 分量得分}}
# 全市场基线
total_scored: int # 参与评分的股票总数
score_mean: float # 全市场 HotScore 均值
score_std: float # 全市场 HotScore 标准差
score_threshold_85_count: int # HotScore ≥ 85 的股票数
score_threshold_70_count: int # HotScore ≥ 70 的股票数
@dataclass
class MarketState:
"""快照时刻的市场状态（用于分段分析）"""
minutes_from_open: int # 距开盘几分钟
session: str # "morning" | "afternoon"
market_up_ratio: float # 全市场上涨比例
market_limit_up_count: int # 全市场涨停家数
market_sentiment: float # 情绪指数 0-100
8.3.2 结果回填（T+N 时刻回填）
@dataclass
class OutcomeRecord:
"""针对快照中每只股票，N 分钟后的真实结果"""
symbol: str
snapshot_id: str # 关联到哪个快照
hotscore_at_snapshot: float # 快照时的 HotScore
# ---- 价格表现 ----
price_at_snapshot: float # 快照时价格
price_after_5m: float # 5 分钟后价格
price_after_10m: float # 10 分钟后价格
price_after_15m: float # 15 分钟后价格
price_after_30m: float # 30 分钟后价格
return_5m: float # 5 分钟收益率 %
return_10m: float # 10 分钟收益率 %
return_15m: float # 15 分钟收益率 %
return_30m: float # 30 分钟收益率 %
max_return_30m: float # 30 分钟内最高收益 %
max_drawdown_30m: float # 30 分钟内最大回撤 %
# ---- 事件标记（bool）----
went_up_10m: bool # 10 分钟后是否上涨
hit_limit_up_30m: bool # 30 分钟内是否涨停
entered_amount_top20_30m: bool # 30 分钟内是否进入成交额 Top20
entered_longhu_today: bool # 当日是否进入龙虎榜
became_sector_leader_30m: bool # 30 分钟内是否成为板块龙头
# ---- HotScore 后续变化 ----
hotscore_after_10m: float # 10 分钟后 HotScore
hotscore_delta_10m: float # HotScore 变化量
stayed_in_top20_10m: bool # 10 分钟后是否仍在 Top20
8.3.3 评估记录（汇总一个快照的整体表现）
@dataclass
class EvalRecord:
"""一个快照的评估汇总（快照 + 全部结果 → 一条评估记录）"""
snapshot_id: str
timestamp: datetime
horizon_minutes: int # 评估时间窗口（5/10/15/30）
top_n: int # Top N 大小（默认 20）
# ---- 涨跌统计 ----
up_count: int # 上涨股票数
down_count: int # 下跌股票数
flat_count: int # 平盘股票数
up_ratio: float # 上涨比例 = up_count / top_n
# ---- 收益统计 ----
avg_return: float # 平均收益率 %
median_return: float # 中位数收益率 %
max_return: float # 最大收益率 %
min_return: float # 最小收益率 %
return_std: float # 收益率标准差
positive_avg_return: float # 上涨股票的平均涨幅 %
negative_avg_return: float # 下跌股票的平均跌幅 %
# ---- 事件统计 ----
limit_up_count: int # 涨停数量
amount_top20_count: int # 进入成交额 Top20 的数量
longhu_count: int # 进入龙虎榜的数量
sector_leader_count: int # 成为板块龙头的数量
# ---- Precision / Recall ----
precision_up: float # 精确率（选出的里面，上涨的比例）
recall_up: float # 召回率（所有上涨的里面，被选出的比例）
f1_up: float # F1 分数
precision_limit_up: float # 涨停精确率
recall_limit_up: float # 涨停召回率
# ---- 对比基线 ----
baseline_avg_return: float # 基线（全市场平均收益）
excess_return: float # 超额收益 = avg_return - baseline
baseline_up_ratio: float # 基线上涨比例
excess_up_ratio: float # 超额上涨比例
# ---- 市场状态 ----
market_state: MarketState
8.4 评估引擎核心逻辑
class EvalEngine:
"""评估引擎 —— 系统的审判官"""
HORIZONS = [5, 10, 15, 30] # 评估时间窗口（分钟）
TOP_N_LIST = [10, 20, 50] # 不同 Top N 粒度
def __init__(self, storage: "EvalStorage"):
self.storage = storage
# 待回填的快照队列：{snapshot_id: (EvalSnapshot, scheduled_time)}
self.pending_snapshots: dict[str, tuple[EvalSnapshot, datetime]] = {}
# =====================================================
# 第一步：每分钟拍快照
# =====================================================
async def take_snapshot(
self,
scores: dict[str, HotScoreResult],
features: dict[str, FeatureSnapshot],
market_state: MarketState,
):
"""
每分钟由主循环调用。
拍下当前 HotScore 排行，存入待回填队列。
"""
now = datetime.now()
snapshot_id = now.strftime("%Y-%m-%d_%H%M")
sorted_scores = sorted(
scores.values(), key=lambda r: r.score, reverse=True
)
for top_n in self.TOP_N_LIST:
top_items = sorted_scores[:top_n]
snapshot = EvalSnapshot(
snapshot_id=f"{snapshot_id}_top{top_n}",
timestamp=now,
market_state=market_state,
top_symbols=[r.symbol for r in top_items],
top_scores={r.symbol: r.score for r in top_items},
top_components={r.symbol: r.components for r in top_items},
total_scored=len(scores),
score_mean=np.mean([r.score for r in scores.values()]),
score_std=np.std([r.score for r in scores.values()]),
score_threshold_85_count=sum(
1 for r in scores.values() if r.score >= 85
),
score_threshold_70_count=sum(
1 for r in scores.values() if r.score >= 70
),
)
# 持久化快照
await self.storage.save_snapshot(snapshot)
# 对每个时间窗口，注册回填任务
for horizon in self.HORIZONS:
backfill_time = now + timedelta(minutes=horizon)
task_id = f"{snapshot.snapshot_id}_h{horizon}"
self.pending_snapshots[task_id] = (snapshot, backfill_time)
# =====================================================
# 第二步：N 分钟后回填结果
# =====================================================
async def backfill_outcomes(
self,
current_prices: dict[str, float],
current_amounts: dict[str, float],
current_scores: dict[str, HotScoreResult],
limit_up_set: set[str],
longhu_set: set[str],
sector_leaders: dict[str, str],
amount_top20: set[str],
):
"""
每分钟由主循环调用。
检查哪些快照已到回填时间，回填真实结果并生成 EvalRecord。
"""
now = datetime.now()
completed = []
for task_id, (snapshot, backfill_time) in self.pending_snapshots.items():
if now < backfill_time:
continue
horizon = int(task_id.split("_h")[-1])
# 为快照中的每只股票计算 OutcomeRecord
outcomes = []
for symbol in snapshot.top_symbols:
outcome = self._build_outcome(
symbol=symbol,
snapshot=snapshot,
horizon=horizon,
current_price=current_prices.get(symbol),
current_amount=current_amounts.get(symbol, 0),
current_score=current_scores.get(symbol),
is_limit_up=symbol in limit_up_set,
is_longhu=symbol in longhu_set,
is_sector_leader=symbol in [
v for v in sector_leaders.values()
],
is_amount_top20=symbol in amount_top20,
)
outcomes.append(outcome)
# 汇总为 EvalRecord
eval_record = self._aggregate_eval(
snapshot, outcomes, horizon, current_prices
)
await self.storage.save_eval_record(eval_record)
await self.storage.save_outcomes(outcomes)
completed.append(task_id)
# 实时推送评估结果到前端
await self._broadcast_eval(eval_record)
for task_id in completed:
del self.pending_snapshots[task_id]
# =====================================================
# 第三步：汇总计算
# =====================================================
def _aggregate_eval(
self,
snapshot: EvalSnapshot,
outcomes: list[OutcomeRecord],
horizon: int,
all_prices: dict[str, float],
) -> EvalRecord:
"""将 N 个 OutcomeRecord 汇总为一条 EvalRecord"""
returns = [self._get_return(o, horizon) for o in outcomes]
top_n = len(outcomes)
up_count = sum(1 for r in returns if r > 0)
down_count = sum(1 for r in returns if r < 0)
flat_count = top_n - up_count - down_count
limit_up_count = sum(1 for o in outcomes if o.hit_limit_up_30m)
amount_top20_count = sum(
1 for o in outcomes if o.entered_amount_top20_30m
)
longhu_count = sum(1 for o in outcomes if o.entered_longhu_today)
sector_leader_count = sum(
1 for o in outcomes if o.became_sector_leader_30m
)
# ---- Precision / Recall ----
precision_up = up_count / max(top_n, 1)
# Recall: 全市场同期上涨股票中，被 Top N 选中的比例
# 需要全市场涨跌数据
market_up_symbols = self._get_market_up_symbols(all_prices, horizon)
recall_up = (
up_count / max(len(market_up_symbols), 1)
if market_up_symbols else 0
)
f1_up = (
2 * precision_up * recall_up / max(precision_up + recall_up, 1e-9)
)
# 涨停的 Precision / Recall
total_limit_up = len(self._get_market_limit_up(horizon))
precision_limit_up = limit_up_count / max(top_n, 1)
recall_limit_up = (
limit_up_count / max(total_limit_up, 1)
if total_limit_up else 0
)
# ---- 基线对比 ----
baseline_returns = self._get_market_avg_return(all_prices, horizon)
baseline_avg = np.mean(baseline_returns) if baseline_returns else 0
baseline_up_ratio = (
sum(1 for r in baseline_returns if r > 0)
/ max(len(baseline_returns), 1)
)
avg_return = np.mean(returns) if returns else 0
positive_returns = [r for r in returns if r > 0]
negative_returns = [r for r in returns if r < 0]
return EvalRecord(
snapshot_id=snapshot.snapshot_id,
timestamp=snapshot.timestamp,
horizon_minutes=horizon,
top_n=top_n,
up_count=up_count,
down_count=down_count,
flat_count=flat_count,
up_ratio=up_count / max(top_n, 1),
avg_return=avg_return,
median_return=float(np.median(returns)) if returns else 0,
max_return=max(returns) if returns else 0,
min_return=min(returns) if returns else 0,
return_std=float(np.std(returns)) if returns else 0,
positive_avg_return=(
np.mean(positive_returns) if positive_returns else 0
),
negative_avg_return=(
np.mean(negative_returns) if negative_returns else 0
),
limit_up_count=limit_up_count,
amount_top20_count=amount_top20_count,
longhu_count=longhu_count,
sector_leader_count=sector_leader_count,
precision_up=precision_up,
recall_up=recall_up,
f1_up=f1_up,
precision_limit_up=precision_limit_up,
recall_limit_up=recall_limit_up,
baseline_avg_return=baseline_avg,
excess_return=avg_return - baseline_avg,
baseline_up_ratio=baseline_up_ratio,
excess_up_ratio=up_count / max(top_n, 1) - baseline_up_ratio,
market_state=snapshot.market_state,
)
8.5 Precision / Recall 详解
传统的 Precision/Recall 框架需要明确"正样本"定义。在这个系统中有多个维度：
维度一：上涨 Precision（最基本）
实际上涨 实际未涨
HotScore Top20 选中 TP FP ← Precision = TP/(TP+FP)
HotScore Top20 未选 FN TN ← Recall = TP/(TP+FN)
- Precision（精确率）：我选出的 20 只里，真正上涨的比例。越高说明"选出来的靠谱"。
- Recall（召回率）：全市场上涨的股票里，被我选中的比例。越高说明"没漏掉好的"。
对于热点侦测系统，Precision 远比 Recall 重要：
我们不需要找到所有上涨的股票（那是全市场选股的事），
我们只需要确保选出来的大概率是对的。
维度二：涨停 Precision
问题：Top20 中有几只后来涨停了？
Precision_涨停 = (Top20中后来涨停的) / 20
Recall_涨停 = (Top20中后来涨停的) / (全市场当日涨停总数)
维度三：成交额 Top20 Precision
问题：HotScore Top20 与成交额 Top20 的重合度？
Overlap = |HotScore_Top20 ∩ Amount_Top20| / 20
维度四：多阈值 Precision
不只看 Top20，还要看不同 HotScore 阈值下的精确率：
| 信号条件 | 选出数量 | 10分钟后上涨数 | Precision | Avg Return |
|---|---|---|---|---|
| Score ≥ 90 | ~5 | ? | ? | ? |
| Score ≥ 85 | ~15 | ? | ? | ? |
| Score ≥ 80 | ~30 | ? | ? | ? |
| Score ≥ 70 | ~80 | ? | ? | ? |
| Top 10 | 10 | ? | ? | ? |
| Top 20 | 20 | ? | ? | ? |
| Top 50 | 50 | ? | ? | ? |
class PrecisionByThreshold:
"""按不同阈值计算 Precision，找到最优 threshold"""
SCORE_THRESHOLDS = [90, 85, 80, 75, 70]
TOPN_THRESHOLDS = [10, 20, 50]
def compute(
self,
all_scores: dict[str, float],
future_returns: dict[str, float],
horizon: int,
) -> list[ThresholdResult]:
results = []
# 按分数阈值
for threshold in self.SCORE_THRESHOLDS:
selected = {s for s, score in all_scores.items() if score >= threshold}
results.append(self._evaluate_set(
name=f"score≥{threshold}",
selected=selected,
future_returns=future_returns,
horizon=horizon,
))
# 按 Top N
sorted_symbols = sorted(
all_scores, key=all_scores.get, reverse=True
)
for n in self.TOPN_THRESHOLDS:
selected = set(sorted_symbols[:n])
results.append(self._evaluate_set(
name=f"top{n}",
selected=selected,
future_returns=future_returns,
horizon=horizon,
))
return results
def _evaluate_set(
self, name: str, selected: set, future_returns: dict, horizon: int
) -> "ThresholdResult":
returns = [future_returns.get(s, 0) for s in selected]
up_count = sum(1 for r in returns if r > 0)
return ThresholdResult(
name=name,
selected_count=len(selected),
up_count=up_count,
precision=up_count / max(len(selected), 1),
avg_return=np.mean(returns) if returns else 0,
median_return=float(np.median(returns)) if returns else 0,
horizon=horizon,
)
@dataclass
class ThresholdResult:
name: str
selected_count: int
up_count: int
precision: float
avg_return: float
median_return: float
horizon: int
8.6 对比基线
HotScore 的价值必须通过对比基线来证明。设计 3 种基线：
class BaselineComparator:
"""
基线对比：如果 HotScore Top20 跑不赢基线，那整个系统没有价值。
"""
def compute_baselines(
self,
all_prices: dict[str, PriceHistory],
hotscore_top20: list[str],
horizon: int,
) -> dict[str, BaselineResult]:
# 基线1：随机选 20 只
random_20 = random.sample(list(all_prices.keys()), 20)
baseline_random = self._eval_portfolio(random_20, all_prices, horizon)
# 基线2：成交额 Top20（市场公认的活跃股）
amount_sorted = sorted(
all_prices, key=lambda s: all_prices[s].current_amount, reverse=True
)
baseline_amount = self._eval_portfolio(
amount_sorted[:20], all_prices, horizon
)
# 基线3：涨幅 Top20（追涨策略）
return_sorted = sorted(
all_prices,
key=lambda s: all_prices[s].intraday_return, reverse=True,
)
baseline_return = self._eval_portfolio(
return_sorted[:20], all_prices, horizon
)
# HotScore Top20 自身
hotscore_result = self._eval_portfolio(
hotscore_top20, all_prices, horizon
)
return {
"hotscore_top20": hotscore_result,
"random_20": baseline_random,
"amount_top20": baseline_amount,
"return_top20": baseline_return,
}
def _eval_portfolio(
self, symbols: list[str], prices: dict, horizon: int
) -> "BaselineResult":
returns = []
for s in symbols:
p = prices.get(s)
if p:
future_price = p.get_price_after(horizon)
ret = (future_price - p.current_price) / p.current_price * 100
returns.append(ret)
up_count = sum(1 for r in returns if r > 0)
return BaselineResult(
avg_return=np.mean(returns) if returns else 0,
median_return=float(np.median(returns)) if returns else 0,
up_ratio=up_count / max(len(returns), 1),
up_count=up_count,
total=len(returns),
)
@dataclass
class BaselineResult:
avg_return: float
median_return: float
up_ratio: float
up_count: int
total: int
每日报告输出示例：
═══════════════════════════════════════════════════════
2026-07-14 评估报告 | 10分钟窗口 | Top20
═══════════════════════════════════════════════════════
策略 平均收益 上涨比例 中位数收益
─────────────────────────────────────────────────
HotScore Top20 +1.23% 75.0% +0.87% ★
成交额 Top20 +0.45% 55.0% +0.21%
涨幅 Top20 +0.12% 50.0% -0.05%
随机 20只 +0.18% 48.0% +0.03%
超额收益（vs 随机）: +1.05% ✓ 有效
超额收益（vs 成交额）: +0.78% ✓ 有效
如果 HotScore Top20 跑不赢"随机选 20 只"，说明整个评分体系无效，需要重新审视权重。
8.7 分时段分析
市场不同时段的特征截然不同，必须分段评估：
class SessionAnalyzer:
"""按交易时段分段评估 HotScore 表现"""
SESSIONS = {
"open_15min": (0, 15), # 开盘 15 分钟（最混乱）
"morning_mid": (15, 60), # 上午中段
"morning_late": (60, 120), # 上午尾段
"afternoon_early": (120, 180), # 下午前段
"afternoon_late": (180, 210), # 下午中段
"close_30min": (210, 240), # 尾盘 30 分钟
}
def analyze_by_session(
self, eval_records: list[EvalRecord]
) -> dict[str, SessionStats]:
"""按时段聚合评估指标"""
session_records: dict[str, list] = {s: [] for s in self.SESSIONS}
for record in eval_records:
minutes = record.market_state.minutes_from_open
for session_name, (start, end) in self.SESSIONS.items():
if start <= minutes < end:
session_records[session_name].append(record)
break
return {
name: self._aggregate_session(records)
for name, records in session_records.items()
if records
}
def _aggregate_session(self, records: list[EvalRecord]) -> "SessionStats":
return SessionStats(
sample_count=len(records),
avg_precision=np.mean([r.precision_up for r in records]),
avg_return=np.mean([r.avg_return for r in records]),
avg_excess_return=np.mean([r.excess_return for r in records]),
avg_up_ratio=np.mean([r.up_ratio for r in records]),
avg_limit_up_count=np.mean([r.limit_up_count for r in records]),
best_precision=max(r.precision_up for r in records),
worst_precision=min(r.precision_up for r in records),
)
@dataclass
class SessionStats:
sample_count: int
avg_precision: float
avg_return: float
avg_excess_return: float
avg_up_ratio: float
avg_limit_up_count: float
best_precision: float
worst_precision: float
输出示例：
时段 样本数 Precision 平均收益 超额收益 涨停命中
──────────────────────────────────────────────────────────
开盘15分钟 15 82% +1.8% +1.5% 2.1
上午中段 45 71% +1.2% +0.9% 1.3
上午尾段 60 65% +0.8% +0.5% 0.8
下午前段 60 68% +0.9% +0.6% 0.9
下午中段 30 58% +0.4% +0.1% 0.4
尾盘30分钟 30 52% +0.2% -0.1% 0.2
→ 结论：HotScore 在开盘 15 分钟效果最好，尾盘几乎失效
→ 行动：考虑尾盘降低 HotScore 权重或停止推送
8.8 HotScore 各分量的 IC 分析
IC（Information Coefficient）衡量每个特征分量与未来收益的相关性，是调整权重的定量依据：
class FeatureICAnalyzer:
"""计算每个 HotScore 分量与未来收益的 IC"""
COMPONENTS = [
"volume_surge", "active_buy", "sector_linkage",
"turnover", "seal_quality",
]
def compute_daily_ic(
self,
snapshots: list[EvalSnapshot],
outcomes: list[OutcomeRecord],
horizon: int = 10,
) -> dict[str, float]:
"""
计算每个分量的截面 IC（Spearman Rank Correlation）
IC > 0.05 → 该分量有预测价值
IC > 0.10 → 该分量是强信号
IC < 0.02 → 该分量基本无用，可降低权重
"""
ic_results = {}
for component in self.COMPONENTS:
scores_list = []
returns_list = []
for snapshot in snapshots:
for symbol in snapshot.top_symbols:
comp_score = snapshot.top_components.get(
symbol, {}
).get(component, 0)
outcome = self._find_outcome(symbol, snapshot.snapshot_id, outcomes)
if outcome:
ret = self._get_return(outcome, horizon)
scores_list.append(comp_score)
returns_list.append(ret)
if len(scores_list) >= 30:
ic, _ = spearmanr(scores_list, returns_list)
ic_results[component] = round(ic, 4)
else:
ic_results[component] = None
return ic_results
def suggest_weight_adjustment(
self, ic_values: dict[str, float]
) -> dict[str, float]:
"""基于 IC 建议权重调整"""
valid = {k: v for k, v in ic_values.items() if v is not None and v > 0}
if not valid:
return {}
total_ic = sum(valid.values())
suggested = {k: round(v / total_ic, 2) for k, v in valid.items()}
return suggested
输出示例：
HotScore 分量 IC 分析（10分钟窗口，最近 5 个交易日）
─────────────────────────────────────────────────
分量 IC值 当前权重 建议权重 结论
volume_surge 0.128 0.35 0.32 ✓ 有效，核心信号
active_buy 0.095 0.25 0.24 ✓ 有效
sector_linkage 0.082 0.20 0.21 ✓ 有效
turnover 0.018 0.10 0.05 ⚠ 偏弱，降低权重
seal_quality 0.071 0.10 0.18 ↑ 被低估，提高权重
→ 建议：turnover 权重从 0.10 降至 0.05，seal_quality 从 0.10 升至 0.18
8.9 每日评估报告
@dataclass
class DailyEvalReport:
"""每日收盘后生成的完整评估报告"""
date: str
trading_minutes: int # 当日交易分钟数
# ---- 信号统计 ----
total_snapshots: int # 拍了多少个快照
total_unique_symbols: int # 总共出现过多少只不同的股票
# ---- 按时间窗口的核心指标 ----
metrics_by_horizon: dict[int, HorizonMetrics] # {5: ..., 10: ..., 15: ..., 30: ...}
# ---- 基线对比 ----
baseline_comparison: dict[int, dict[str, BaselineResult]]
# ---- 分时段分析 ----
session_analysis: dict[str, SessionStats]
# ---- IC 分析 ----
component_ic: dict[str, float]
weight_suggestions: dict[str, float]
# ---- 最佳/最差案例 ----
best_snapshots: list[str] # 表现最好的 5 个快照 ID
worst_snapshots: list[str] # 表现最差的 5 个快照 ID
# ---- 趋势（与过去 N 日对比）----
precision_trend_7d: list[float] # 最近 7 日 Precision 趋势
excess_return_trend_7d: list[float] # 最近 7 日超额收益趋势
@dataclass
class HorizonMetrics:
"""单个时间窗口的汇总指标"""
horizon: int
# Precision / Recall 均值
avg_precision: float
avg_recall: float
avg_f1: float
# 收益
avg_return: float
median_return: float
avg_excess_return: float
# 涨跌统计
avg_up_ratio: float
total_limit_up_hits: int
total_amount_top20_hits: int
total_longhu_hits: int
# 最好 / 最差快照
best_precision: float
worst_precision: float
precision_std: float
# 多阈值分析
threshold_results: list[ThresholdResult]
8.10 评估数据存储
-- 评估快照表（每分钟一条 × 每个 Top N 粒度）
-- 路径: data/hotspot/eval/snapshots/date={yyyy-mm-dd}/data.parquet
CREATE TABLE eval_snapshots (
snapshot_id VARCHAR PRIMARY KEY,
timestamp TIMESTAMP,
top_n INTEGER,
top_symbols JSON, -- ["600519", "300750", ...]
top_scores JSON, -- {"600519": 92.5, ...}
top_components JSON, -- {"600519": {"volume_surge": 88, ...}, ...}
total_scored INTEGER,
score_mean DOUBLE,
score_std DOUBLE,
-- 市场状态
minutes_from_open INTEGER,
session VARCHAR,
market_up_ratio DOUBLE,
market_limit_up_count INTEGER,
market_sentiment DOUBLE
);
-- 结果回填表（每只股票 × 每个快照 × 每个时间窗口一条）
-- 路径: data/hotspot/eval/outcomes/date={yyyy-mm-dd}/data.parquet
CREATE TABLE eval_outcomes (
snapshot_id VARCHAR,
symbol VARCHAR,
horizon_minutes INTEGER,
hotscore_at_snapshot DOUBLE,
price_at_snapshot DOUBLE,
price_after DOUBLE,
return_pct DOUBLE,
max_return_pct DOUBLE,
max_drawdown_pct DOUBLE,
went_up BOOLEAN,
hit_limit_up BOOLEAN,
entered_amount_top20 BOOLEAN,
entered_longhu BOOLEAN,
became_sector_leader BOOLEAN,
hotscore_after DOUBLE,
hotscore_delta DOUBLE,
stayed_in_topn BOOLEAN,
PRIMARY KEY (snapshot_id, symbol, horizon_minutes)
);
-- 评估汇总表（每个快照 × 每个时间窗口一条）
-- 路径: data/hotspot/eval/records/date={yyyy-mm-dd}/data.parquet
CREATE TABLE eval_records (
snapshot_id VARCHAR,
horizon_minutes INTEGER,
top_n INTEGER,
timestamp TIMESTAMP,
-- 涨跌
up_count INTEGER,
down_count INTEGER,
up_ratio DOUBLE,
-- 收益
avg_return DOUBLE,
median_return DOUBLE,
max_return DOUBLE,
min_return DOUBLE,
return_std DOUBLE,
-- 事件
limit_up_count INTEGER,
amount_top20_count INTEGER,
longhu_count INTEGER,
sector_leader_count INTEGER,
-- Precision / Recall
precision_up DOUBLE,
recall_up DOUBLE,
f1_up DOUBLE,
precision_limit_up DOUBLE,
recall_limit_up DOUBLE,
-- 基线对比
baseline_avg_return DOUBLE,
excess_return DOUBLE,
-- 市场状态
minutes_from_open INTEGER,
session VARCHAR,
PRIMARY KEY (snapshot_id, horizon_minutes)
);
-- 每日评估报告表
CREATE TABLE eval_daily_report (
date DATE,
horizon_minutes INTEGER,
top_n INTEGER,
total_snapshots INTEGER,
avg_precision DOUBLE,
avg_recall DOUBLE,
avg_return DOUBLE,
avg_excess_return DOUBLE,
avg_up_ratio DOUBLE,
total_limit_up_hits INTEGER,
total_longhu_hits INTEGER,
best_precision DOUBLE,
worst_precision DOUBLE,
component_ic JSON, -- {"volume_surge": 0.128, ...}
weight_suggestions JSON,
precision_trend_7d JSON, -- [0.72, 0.68, 0.71, ...]
PRIMARY KEY (date, horizon_minutes, top_n)
);
8.11 评估驱动的权重自动调优
评估层不只是"看看数字"，它直接驱动 HotScore 权重的迭代：
Week 1: 默认权重（规划书推荐）
↓ 收集 5 个交易日的评估数据
↓
Week 2: IC 分析 → 发现 turnover IC 只有 0.018，seal_quality IC 有 0.071
↓ 权重调整建议：降 turnover，升 seal_quality
↓
Week 3: 应用新权重，继续评估
↓ 对比 Week 2 vs Week 3 的 Precision 和超额收益
↓
Week 4: 如果改善 → 固化；如果恶化 → 回滚
每周循环，永不停止。
class WeightOptimizer:
"""基于评估数据的权重自动调优"""
def optimize(
self,
recent_eval: list[DailyEvalReport],
current_weights: dict[str, float],
min_ic_threshold: float = 0.02,
) -> WeightUpdate:
# 1. 计算近 5 日平均 IC
avg_ic = self._average_ic(recent_eval)
# 2. IC 过低的分量降权
adjusted = {}
for comp, ic in avg_ic.items():
if ic is None or ic < min_ic_threshold:
adjusted[comp] = current_weights[comp] * 0.5 # 砍半
else:
adjusted[comp] = ic # IC 加权
# 3. 归一化使权重和为 1
total = sum(adjusted.values())
new_weights = {k: round(v / total, 3) for k, v in adjusted.items()}
# 4. 与当前权重对比，变化过大则限制步长
for comp in new_weights:
delta = new_weights[comp] - current_weights[comp]
if abs(delta) > 0.10: # 单次调整不超过 10%
new_weights[comp] = current_weights[comp] + 0.10 * np.sign(delta)
return WeightUpdate(
old_weights=current_weights,
new_weights=new_weights,
ic_values=avg_ic,
reason="weekly_ic_optimization",
)
8.12 评估层在架构中的位置提升
原架构将评估放在最末层。现在把它提升到与 HotScore 引擎同级的核心位置：
采集层 → 特征层 → HotScore 引擎 ←──→ 评估引擎（审判官）
↑ │
│ │
└── 权重调整 ←────────┘
在主循环中，评估引擎不是"事后总结"，而是伴随运行：
async def main_loop():
"""主循环（每分钟执行一次）"""
while is_trading_time():
# 1. 特征计算
features = await feature_engine.compute()
# 2. HotScore 计算
scores = hotscore_engine.compute_all(features)
# 3. 评估引擎：拍快照（记录当前 Top20）
await eval_engine.take_snapshot(scores, features, market_state)
# 4. 评估引擎：回填之前的快照（用当前事实）
await eval_engine.backfill_outcomes(
current_prices, current_amounts, scores,
limit_up_set, longhu_set, sector_leaders, amount_top20,
)
# 5. 实时推送排行 + 评估数据到前端
await ws_manager.broadcast_ranking(scores)
await ws_manager.broadcast_eval(eval_engine.latest_eval)
await asyncio.sleep(60)
# 收盘后
daily_report = eval_engine.generate_daily_report()
await eval_storage.save_daily_report(daily_report)
# 每周五自动运行权重优化建议
if is_friday():
suggestion = weight_optimizer.optimize(
recent_eval=eval_storage.load_recent(days=5),
current_weights=hotscore_engine.weights,
)
logger.info(f"权重优化建议: {suggestion}")
8.13 评估看板设计
在前端增加独立的评估看板 Tab：
┌──────────────────────────────────────────────────────────────────┐
│ [热点排行] [板块联动] [告警] [★ 评估看板] │
├──────────────────────────────────────────────────────────────────┤
│ │
│ ┌─ 实时 Precision 仪表盘 ──────────────────────────────────┐ │
│ │ │ │
│ │ 当前 Precision (10min) [====████████====] 72% │ │
│ │ 当前超额收益 (10min) +0.85% vs 随机 +0.12% │ │
│ │ 今日涨停命中 3 / 18 (16.7%) │ │
│ │ 今日龙虎榜命中 1 / 8 (12.5%) │ │
│ │ │ │
│ └───────────────────────────────────────────────────────────┘ │
│ │
│ ┌─ Precision 时序图 ─────────┐ ┌─ 超额收益时序图 ──────────┐ │
│ │ Y: Precision % │ │ Y: 超额收益 % │ │
│ │ X: 09:30 ────── 15:00 │ │ X: 09:30 ────── 15:00 │ │
│ │ ···· 70% 基准线 │ │ ···· 0% 基准线 │ │
│ │ ▁▃▅█▇▆▅▃▂▁▃▅▇█▆▅▃ │ │ ▂▅█▇▅▃▁▂▃▅▇█▆▅▃▂ │ │
│ └────────────────────────────┘ └───────────────────────────┘ │
│ │
│ ┌─ 多阈值 Precision 对比 ───────────────────────────────────┐ │
│ │ Score≥90 ████████████████████████████ 85% (avg +2.1%) │ │
│ │ Score≥85 ██████████████████████████ 78% (avg +1.5%) │ │
│ │ Top 10 █████████████████████████ 75% (avg +1.3%) │ │
│ │ Top 20 ████████████████████████ 72% (avg +1.1%) │ │
│ │ Score≥70 ██████████████████ 62% (avg +0.6%) │ │
│ │ Top 50 ████████████████ 58% (avg +0.3%) │ │
│ └───────────────────────────────────────────────────────────┘ │
│ │
│ ┌─ 基线对比 ──────────────────┐ ┌─ 分时段 Precision ───────┐ │
│ │ HotScore Top20 ████ 1.2% │ │ 开盘15min ████████ 82% │ │
│ │ 成交额 Top20 ██ 0.5% │ │ 上午中段 ██████ 71% │ │
│ │ 涨幅 Top20 █ 0.1% │ │ 上午尾段 █████ 65% │ │
│ │ 随机 20 只 █ 0.2% │ │ 下午前段 ██████ 68% │ │
│ └────────────────────────────┘ │ 下午中段 ████ 58% │ │
│ │ 尾盘30min ███ 52% │ │
│ └──────────────────────────┘ │
│ │
│ ┌─ IC 分析（最近5日）─────────────────────────────────────────┐ │
│ │ volume_surge ▓▓▓▓▓▓▓▓▓▓▓▓▓ 0.128 (权重 0.35 → 0.32) │ │
│ │ active_buy ▓▓▓▓▓▓▓▓▓▓ 0.095 (权重 0.25 → 0.24) │ │
│ │ sector_link ▓▓▓▓▓▓▓▓ 0.082 (权重 0.20 → 0.21) │ │
│ │ seal_quality ▓▓▓▓▓▓▓ 0.071 (权重 0.10 → 0.18) │ │
│ │ turnover ▓▓ 0.018 (权重 0.10 → 0.05) │ │
│ └─────────────────────────────────────────────────────────────┘ │
└──────────────────────────────────────────────────────────────────┘
8.14 评估层验收标准
| 阶段 | 指标 | 及格线 | 优秀线 | 失败线（需要停下来反思） |
|---|---|---|---|---|
| MVP | Top20 10min Precision | ≥55% | ≥70% | <45%（不如随机） |
| MVP | Top20 10min 超额收益 | >0% | >+0.5% | <0%（跑输随机） |
| MVP | Top20 10min 上涨比例 | >全市场均值 | >均值+15% | <全市场均值 |
| 成熟 | 涨停 Recall (30min) | ≥10% | ≥20% | <5% |
| 成熟 | IC (volume_surge) | >0.05 | >0.10 | <0.02 |
| 成熟 | 连续5日 Precision 稳定性 | std<10% | std<5% | std>20% |
关键决策点：
- 如果 Precision 连续 3 个交易日 < 45% → 暂停系统，检查数据质量和特征计算
- 如果超额收益连续 5 个交易日 < 0 → 回滚到上一版权重，排查原因
- 只有 Precision 稳定 ≥ 55% 且超额 > 0 至少 10 个交易日 → 才有资格开始训练 AI 模型
九、API 与前端展示层详细设计
9.1 API 路由设计
热点排行 API
| 方法 | 路径 | 说明 | 响应频率 |
|---|---|---|---|
| GET | /api/hotspot/ranking |
当前热点排行 Top N | 实时 |
| GET | /api/hotspot/detail/{symbol} |
单只股票的热点详情（特征 + 评分明细） | 实时 |
| GET | /api/hotspot/history/{symbol} |
单只股票的 HotScore 历史曲线 | 分钟级 |
| GET | /api/hotspot/surging |
急速升温股票列表 | 实时 |
板块联动 API
| 方法 | 路径 | 说明 |
|---|---|---|
| GET | /api/sector/ranking |
板块热度排行 |
| GET | /api/sector/detail/{sector} |
板块详情（成分股 + 龙头 + 指标） |
| GET | /api/sector/resonance |
当前板块共振事件 |
告警 API
| 方法 | 路径 | 说明 |
|---|---|---|
| GET | /api/alert/recent |
最近告警列表 |
| WS | /ws/alerts |
实时告警推送 |
评估 API（核心）
| 方法 | 路径 | 说明 |
|---|---|---|
| GET | /api/eval/realtime |
当前实时 Precision / 超额收益 / 涨停命中 |
| GET | /api/eval/precision_series/{date} |
当日 Precision 时序数据（绘图用） |
| GET | /api/eval/threshold_analysis |
多阈值 Precision 分析（Score≥90/85/80/70, Top10/20/50） |
| GET | /api/eval/baseline/{date} |
基线对比（HotScore vs 随机/成交额Top/涨幅Top） |
| GET | /api/eval/session/{date} |
分时段分析（开盘/上午/下午/尾盘） |
| GET | /api/eval/ic/{date} |
各分量 IC 分析 + 权重调整建议 |
| GET | /api/eval/daily/{date} |
指定日期完整评估报告 |
| GET | /api/eval/trend |
最近 N 日 Precision / 超额收益趋势 |
| GET | /api/eval/snapshots/{snapshot_id} |
查看某个快照的详细结果 |
系统状态 API
| 方法 | 路径 | 说明 |
|---|---|---|
| GET | /api/system/collectors |
采集器状态（运行 / 延迟 / 错误） |
| GET | /api/system/config |
当前配置（权重等） |
| PUT | /api/system/config/weights |
热更新 HotScore 权重 |
9.2 WebSocket 实时推送
class WebSocketManager:
"""WebSocket 连接管理器"""
def __init__(self):
self.active_connections: list[WebSocket] = []
async def connect(self, websocket: WebSocket):
await websocket.accept()
self.active_connections.append(websocket)
async def broadcast_ranking(self, ranking: list[dict]):
"""广播热点排行更新（每分钟）"""
msg = {"type": "ranking_update", "data": ranking, "ts": now_str()}
await self._broadcast(msg)
async def broadcast_alert(self, alert: Alert):
"""广播告警（实时）"""
msg = {"type": "alert", "data": asdict(alert), "ts": now_str()}
await self._broadcast(msg)
async def _broadcast(self, message: dict):
dead = []
for conn in self.active_connections:
try:
await conn.send_json(message)
except WebSocketDisconnect:
dead.append(conn)
for conn in dead:
self.active_connections.remove(conn)
9.3 前端面板设计（hotspot_dashboard.html）
布局规划：
┌──────────────────────────────────────────────────────────────────┐
│ 顶部状态栏：系统状态 │ 采集延迟 │ 时间 │ 市场情绪指数 │
├────────────────────────────┬─────────────────────────────────────┤
│ │ │
│ 热点排行 Top 20 │ 选中股票详情 │
│ ┌────┬───────┬─────┐ │ ┌─────────────────────────────┐ │
│ │排名│ 代码 │ 分数 │ │ │ HotScore 走势图 │ │
│ ├────┼───────┼─────┤ │ │ (折线图，5秒刷新) │ │
│ │ 1 │600519 │ 92 │ │ ├─────────────────────────────┤ │
│ │ 2 │300750 │ 88 │ │ │ 特征雷达图 │ │
│ │ ...│ ... │ ... │ │ │ (成交/资金/盘口/联动 4维) │ │
│ └────┴───────┴─────┘ │ ├─────────────────────────────┤ │
│ (点击行 → 右侧显示详情) │ │ 分钟成交额柱状图 │ │
│ │ │ 主动买卖比例 │ │
│ │ └─────────────────────────────┘ │
├────────────────────────────┼─────────────────────────────────────┤
│ │ │
│ 板块热度排行 │ 告警流 │
│ ┌──────────┬─────┐ │ ┌─────────────────────────────┐ │
│ │ 板块名 │ 热度 │ │ │ 09:35 600519 成交额爆发 │ │
│ │ 人工智能 │ 85 │ │ │ 09:37 板块共振: AI+算力 │ │
│ │ 半导体 │ 78 │ │ │ 09:42 300750 急速升温 │ │
│ └──────────┴─────┘ │ │ ... │ │
│ │ └─────────────────────────────┘ │
└────────────────────────────┴─────────────────────────────────────┘
十、存储层详细设计
10.1 Redis 数据结构
| Key 模式 | 类型 | 说明 | TTL |
|---|---|---|---|
hotspot:ranking:{date} |
Sorted Set | 热点排行榜 (score=HotScore) | 24h |
hotspot:features:{symbol} |
Hash | 单只股票最新特征快照 | 盘中更新 |
hotspot:score_history:{symbol}:{date} |
List | HotScore 分钟级历史 | 7d |
sector:metrics:{sector} |
Hash | 板块最新指标 | 盘中更新 |
sector:ranking:{date} |
Sorted Set | 板块热度排行 | 24h |
alert:recent |
List (capped) | 最近 100 条告警 | 24h |
system:collector_stats |
Hash | 采集器状态 | 盘中更新 |
config:hotscore_weights |
Hash | HotScore 权重（热更新） | 永久 |
eval:latest:{horizon} |
Hash | 最新一次评估结果 (precision/return等) | 盘中更新 |
eval:precision_series:{date} |
List | 当日 Precision 时序（绘图用） | 7d |
eval:pending_snapshots |
Hash | 待回填的快照队列 | 盘中更新 |
10.2 DuckDB 表设计（历史数据）
-- 特征历史表（按日期 Hive 分区）
-- 路径: data/hotspot/features/date={yyyy-mm-dd}/data.parquet
CREATE TABLE feature_history (
symbol VARCHAR,
timestamp TIMESTAMP,
-- 成交特征
volume_surge_5m DOUBLE,
volume_surge_1m DOUBLE,
volume_ratio DOUBLE,
turnover_rate DOUBLE,
large_order_ratio DOUBLE,
trade_density DOUBLE,
-- 资金主动性
active_buy_amount DOUBLE,
active_sell_amount DOUBLE,
net_active_buy DOUBLE,
active_buy_ratio DOUBLE,
consecutive_buy_count INTEGER,
buy_momentum DOUBLE,
-- 盘口结构
bid_ask_imbalance DOUBLE,
seal_amount DOUBLE,
seal_duration DOUBLE,
cancel_rate DOUBLE,
ask_decay_speed DOUBLE,
spread DOUBLE,
depth_ratio DOUBLE,
-- 联动特征
sector_change_pct DOUBLE,
sector_up_ratio DOUBLE,
sector_limitup_count INTEGER,
sector_volume_surge DOUBLE,
market_sentiment DOUBLE,
sector_rank INTEGER,
-- HotScore
hotscore DOUBLE,
hotscore_level VARCHAR
);
-- 评估相关表见第八章 8.10 节（eval_snapshots / eval_outcomes /
-- eval_records / eval_daily_report 共 4 张表）
-- 告警历史表
CREATE TABLE alert_history (
id INTEGER,
symbol VARCHAR,
rule_name VARCHAR,
level VARCHAR,
description VARCHAR,
timestamp TIMESTAMP,
context JSON
);
10.3 MySQL 表设计（配置与映射）
-- 板块-个股映射表
CREATE TABLE sector_mapping (
id INT AUTO_INCREMENT PRIMARY KEY,
sector_name VARCHAR(50) NOT NULL COMMENT '板块名称',
sector_code VARCHAR(20) NOT NULL COMMENT '板块代码',
sector_type ENUM('concept', 'industry') NOT NULL,
symbol VARCHAR(10) NOT NULL COMMENT '股票代码',
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
UNIQUE KEY uk_sector_symbol (sector_code, symbol),
KEY idx_symbol (symbol),
KEY idx_sector (sector_code)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
-- 关注列表（动态管理）
CREATE TABLE watch_list (
symbol VARCHAR(10) PRIMARY KEY,
source VARCHAR(20) NOT NULL COMMENT '来源: pool/scan/manual',
added_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
is_active TINYINT(1) DEFAULT 1,
reason VARCHAR(200) COMMENT '加入原因'
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
十一、配置文件设计
11.1 settings.yaml
system:
name: "market_hotspot"
log_level: "INFO"
timezone: "Asia/Shanghai"
collector:
tick:
interval_seconds: 3
max_symbols: 500
orderbook:
interval_seconds: 3
batch_size: 80
sector:
interval_seconds: 30
limitup:
interval_seconds: 60
connection_pool:
size: 5
reconnect_delay: 5
max_retries: 3
feature:
buffer_minutes: 30
compute_interval_seconds: 60
large_order_threshold: 500000
hotscore:
weights:
volume_surge: 0.35
active_buy: 0.25
sector_linkage: 0.20
turnover: 0.10
seal_quality: 0.10
thresholds:
strong_hot: 85
potential_hot: 70
alert:
cooldown_minutes: 5
max_alerts_per_minute: 20
model:
enabled: false # MVP 阶段先关闭
label_config: "hotscore_surge_10m"
retrain_schedule: "daily"
min_auc_threshold: 0.65
storage:
redis:
host: "127.0.0.1"
port: 6379
db: 1
duckdb:
data_root: "data/hotspot"
mysql:
host: "192.168.129.1"
port: 3306
database: "quant_data"
username: "claw"
password: "claw"
api:
host: "0.0.0.0"
port: 8001 # 与 stock_pool_app(8000) 错开
ws_ping_interval: 30
trading_hours:
pre_market: "09:15"
morning_open: "09:30"
morning_close: "11:30"
afternoon_open: "13:00"
afternoon_close: "15:00"
post_market: "15:05"
11.2 hotscore_weights.yaml（支持热更新）
version: 1
updated_at: "2026-07-13"
note: "初始权重，来自规划书推荐值"
weights:
volume_surge: 0.35
active_buy: 0.25
sector_linkage: 0.20
turnover: 0.10
seal_quality: 0.10
thresholds:
strong_hot: 85
potential_hot: 70
normalization:
method: "percentile_rank" # percentile_rank | min_max | z_score
十二、主程序启动流程
# main.py 启动流程伪代码
async def main():
# 1. 加载配置
config = load_config("config/settings.yaml")
# 2. 初始化存储
redis = await init_redis(config["storage"]["redis"])
duckdb = init_duckdb(config["storage"]["duckdb"])
# 3. 初始化板块映射
sector_mapping = SectorMapping()
await sector_mapping.refresh_from_eastmoney()
# 4. 初始化各引擎
feature_engine = FeatureEngine(config["feature"])
hotscore_engine = HotScoreEngine()
sector_engine = SectorEngine(sector_mapping)
alert_engine = AlertEngine(config["alert"])
# 5. 初始化采集器
manager = CollectorManager(config["collector"])
manager.register(TickCollector(...))
manager.register(OrderBookCollector(...))
manager.register(SectorCollector(...))
manager.register(LimitUpCollector(...))
# 6. 启动 FastAPI + WebSocket
app = create_fastapi_app(config, engines={...})
server = uvicorn.Server(uvicorn.Config(app, host="0.0.0.0", port=8001))
# 7. 启动主循环
await asyncio.gather(
manager.start_all(), # 数据采集
feature_engine.run(), # 特征计算
hotscore_engine.run(), # HotScore 计算
alert_engine.run(), # 告警检测
server.serve(), # API 服务
)
十三、开发路线图
Phase 1（第1周）：数据基础
| 任务 | 优先级 | 预计工时 |
|---|---|---|
| 项目骨架 + 配置管理 | P0 | 0.5d |
| pytdx 连接池 + 逐笔采集 | P0 | 1.5d |
| 五档盘口采集 | P0 | 1d |
| DataBus 数据总线 | P0 | 0.5d |
| 采集监控 + 日志 | P1 | 0.5d |
| 验收标准 | 盘中稳定采集 300 只股票，丢包率<1%，连续运行 2 个交易日无崩溃 |
Phase 2（第2周）：特征 + HotScore + 评估骨架
| 任务 | 优先级 | 预计工时 |
|---|---|---|
| 时序缓冲区 + 分钟预聚合 | P0 | 1d |
| 成交特征 + 资金特征计算 | P0 | 1d |
| 盘口结构特征计算 | P0 | 0.5d |
| HotScore 引擎 + 截面归一化 | P0 | 1d |
| 评估引擎：快照 + 回填 + EvalRecord | P0 | 1d |
| Redis 排行榜 + API | P1 | 0.5d |
| 验收标准 | HotScore 每分钟刷新 + 每分钟自动拍快照 + 10 分钟后自动回填结果 |
第 2 周结束时必须能回答：Top20 的 10 分钟 Precision 是多少？超额收益是多少？
如果不能回答，不要进入第 3 周。
Phase 3（第3周）：板块联动 + 评估体系完善
| 任务 | 优先级 | 预计工时 |
|---|---|---|
| 板块行情 + 涨停数据采集 | P0 | 1d |
| 板块-个股映射管理 | P0 | 0.5d |
| 板块联动引擎 + 龙头识别 | P0 | 1d |
| 联动特征注入 HotScore | P0 | 0.5d |
| 多阈值 Precision + 基线对比 + 分时段分析 | P0 | 1d |
| IC 分析 + 权重调整建议 | P0 | 0.5d |
| 告警引擎 + WebSocket 推送 | P1 | 0.5d |
| 验收标准 | Precision ≥ 55% 且超额收益 > 0（连续 3 个交易日），否则调整权重后重新验证 |
Phase 4（第4周）：评估看板 + AI（仅当评估达标时）
| 任务 | 优先级 | 预计工时 |
|---|---|---|
| 评估看板前端（Precision 曲线、基线对比、IC 图） | P0 | 1.5d |
| 每日评估报告自动生成 | P0 | 0.5d |
| 权重自动调优（基于 IC） | P1 | 1d |
| 历史数据持久化管道 | P0 | 0.5d |
| LightGBM 训练 + 在线推理（仅当 Precision 稳定 ≥55%） | P2 | 1.5d |
| 验收标准 | 评估看板可用 + 10 个交易日 Precision 趋势稳定 |
Phase 4 的 AI 部分是可选的。 如果 Phase 3 结束时 Precision < 55%，
第 4 周应全力优化 HotScore 权重和特征，而不是上 AI。
AI 在一个"不准"的 HotScore 上训练，只会放大错误。
Phase 5（第5-6周）：优化迭代
- 权重滚动优化（每周基于 IC 微调）
- 如果 Precision 稳定，启动 AI 模型训练
- 评估 AI 模型 vs 纯 HotScore 的增量价值
- 性能优化（采集并发、计算向量化）
- 与 stock_pool_app 集成（观察池打通）
- 移动端告警推送（可选）
十四、依赖清单
# requirements.txt
# Web 框架
fastapi>=0.100.0
uvicorn[standard]>=0.23.0
websockets>=12.0
# 数据源
pytdx>=1.72
akshare>=1.12.0
# 数据处理
pandas>=2.0.0
numpy>=1.24.0
# 存储
redis[hiredis]>=5.0.0
duckdb>=0.10.0
sqlalchemy>=2.0.0
pymysql>=1.1.0
# ML（Phase 4 启用）
lightgbm>=4.0.0
scikit-learn>=1.3.0
# 调度与工具
apscheduler>=3.10.0
pyyaml>=6.0
pydantic>=2.0.0
# 日志
loguru>=0.7.0
暂无评论，快来抢沙发吧~