1521 lines
52 KiB
Python
1521 lines
52 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""
|
||
实时概念异动检测服务（实盘可用）
|
||
|
||
盘中每分钟运行一次，检测概念异动并写入数据库
|
||
|
||
数据流程：
|
||
1. 启动时从 ES 获取概念列表，从 MySQL 获取昨收价
|
||
2. 自动预热：从 ClickHouse 加载当天已有的历史分钟数据
|
||
3. 每分钟增量获取最新分钟数据
|
||
4. 在内存中实时计算概念特征（无前瞻偏差）
|
||
5. 使用规则+ML融合评分检测异动
|
||
6. 异动写入 MySQL
|
||
|
||
特征计算说明（无 Looking Forward）：
|
||
- alpha: 当前时间点的概念超额收益
|
||
- alpha_delta: 使用过去 5 分钟的 alpha 变化
|
||
- amt_ratio: 使用过去 20 分钟的成交额均值
|
||
- rank_pct: 当前时间点所有概念的 alpha 排名
|
||
- limit_up_ratio: 当前时间点的涨停股占比
|
||
|
||
使用方法：
|
||
# 实盘模式（推荐）- 自动预热，不依赖 prepare_data.py
|
||
python realtime_detector.py
|
||
|
||
# 单次检测
|
||
python realtime_detector.py --once
|
||
|
||
# 回补历史异动到数据库（需要 prepare_data.py 生成 parquet）
|
||
python realtime_detector.py --backfill-only
|
||
|
||
# 实盘模式 + 启动时回补历史
|
||
python realtime_detector.py --backfill
|
||
|
||
最小数据量要求：
|
||
- ML 评分需要 seq_len=15 分钟的序列
|
||
- amt_ratio 需要 amt_ma_window=20 分钟的历史
|
||
- 即：开盘后约 35 分钟才能正常工作
|
||
"""
|
||
|
||
import os
|
||
import sys
|
||
import time
|
||
import json
|
||
import argparse
|
||
import schedule
|
||
from datetime import datetime, timedelta
|
||
from pathlib import Path
|
||
from typing import Dict, List, Optional, Tuple, Set
|
||
from collections import defaultdict
|
||
|
||
import numpy as np
|
||
import pandas as pd
|
||
import torch
|
||
from sqlalchemy import create_engine, text
|
||
from elasticsearch import Elasticsearch
|
||
from clickhouse_driver import Client as CHClient
|
||
|
||
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||
|
||
|
||
# ==================== 配置 ====================
|
||
|
||
MYSQL_ENGINE = create_engine(
|
||
"mysql+pymysql://root:Zzl5588161!@192.168.1.5:3306/stock",
|
||
echo=False,
|
||
pool_pre_ping=True,
|
||
pool_recycle=3600,
|
||
)
|
||
|
||
ES_CLIENT = Elasticsearch(['http://127.0.0.1:9200'])
|
||
ES_INDEX = 'concept_library_v3'
|
||
|
||
CLICKHOUSE_CONFIG = {
|
||
'host': '127.0.0.1',
|
||
'port': 9000,
|
||
'user': 'default',
|
||
'password': 'Zzl33818!',
|
||
'database': 'stock'
|
||
}
|
||
|
||
REFERENCE_INDEX = '000001.SH'
|
||
|
||
FEATURES = ['alpha', 'alpha_delta', 'amt_ratio', 'amt_delta', 'rank_pct', 'limit_up_ratio']
|
||
|
||
# 特征计算参数
|
||
FEATURE_CONFIG = {
|
||
'alpha_delta_window': 5,
|
||
'amt_ma_window': 20,
|
||
'limit_up_threshold': 9.8,
|
||
'limit_down_threshold': -9.8,
|
||
}
|
||
|
||
# 检测配置（与 backtest_fast.py 保持一致）
|
||
CONFIG = {
|
||
'seq_len': 15,
|
||
'min_alpha_abs': 0.3,
|
||
'cooldown_minutes': 8,
|
||
'max_alerts_per_minute': 20,
|
||
'clip_value': 10.0,
|
||
# === 融合权重：与 backtest_fast.py 一致 ===
|
||
'rule_weight': 0.5,
|
||
'ml_weight': 0.5,
|
||
# === 触发阈值：与 backtest_fast.py 一致 ===
|
||
'rule_trigger': 65,
|
||
'ml_trigger': 70,
|
||
'fusion_trigger': 45,
|
||
}
|
||
|
||
TRADING_PERIODS = [
|
||
('09:30', '11:30'),
|
||
('13:00', '15:00'),
|
||
]
|
||
|
||
|
||
# ==================== 工具函数 ====================
|
||
|
||
def get_ch_client():
|
||
return CHClient(**CLICKHOUSE_CONFIG)
|
||
|
||
|
||
def code_to_ch_format(code: str) -> str:
|
||
if not code or len(code) != 6 or not code.isdigit():
|
||
return None
|
||
if code.startswith('6'):
|
||
return f"{code}.SH"
|
||
elif code.startswith('0') or code.startswith('3'):
|
||
return f"{code}.SZ"
|
||
else:
|
||
return f"{code}.BJ"
|
||
|
||
|
||
def is_trading_time() -> bool:
|
||
now = datetime.now()
|
||
if now.weekday() >= 5:
|
||
return False
|
||
current_time = now.strftime('%H:%M')
|
||
for start, end in TRADING_PERIODS:
|
||
if start <= current_time <= end:
|
||
return True
|
||
return False
|
||
|
||
|
||
def get_current_trade_date() -> str:
|
||
now = datetime.now()
|
||
if now.hour < 9:
|
||
now = now - timedelta(days=1)
|
||
return now.strftime('%Y-%m-%d')
|
||
|
||
|
||
# ==================== 数据获取 ====================
|
||
|
||
def get_all_concepts() -> List[dict]:
|
||
"""从 ES 获取所有概念"""
|
||
concepts = []
|
||
query = {
|
||
"query": {"match_all": {}},
|
||
"size": 100,
|
||
"_source": ["concept_id", "concept", "stocks"]
|
||
}
|
||
|
||
resp = ES_CLIENT.search(index=ES_INDEX, body=query, scroll='2m')
|
||
scroll_id = resp['_scroll_id']
|
||
hits = resp['hits']['hits']
|
||
|
||
while len(hits) > 0:
|
||
for hit in hits:
|
||
source = hit['_source']
|
||
stocks = []
|
||
if 'stocks' in source and isinstance(source['stocks'], list):
|
||
for stock in source['stocks']:
|
||
if isinstance(stock, dict) and 'code' in stock and stock['code']:
|
||
stocks.append(stock['code'])
|
||
|
||
if stocks:
|
||
concepts.append({
|
||
'concept_id': source.get('concept_id'),
|
||
'concept_name': source.get('concept'),
|
||
'stocks': stocks
|
||
})
|
||
|
||
resp = ES_CLIENT.scroll(scroll_id=scroll_id, scroll='2m')
|
||
scroll_id = resp['_scroll_id']
|
||
hits = resp['hits']['hits']
|
||
|
||
ES_CLIENT.clear_scroll(scroll_id=scroll_id)
|
||
print(f"获取到 {len(concepts)} 个概念")
|
||
return concepts
|
||
|
||
|
||
def get_prev_close(stock_codes: List[str], trade_date: str) -> Dict[str, float]:
|
||
"""获取昨收价（上一交易日的收盘价 F007N）"""
|
||
valid_codes = [c for c in stock_codes if c and len(c) == 6 and c.isdigit()]
|
||
if not valid_codes:
|
||
return {}
|
||
|
||
codes_str = "','".join(valid_codes)
|
||
# 注意：F007N 是"最近成交价"即当日收盘价，F002N 是"昨日收盘价"
|
||
# 我们需要查上一交易日的 F007N（那天的收盘价）作为今天的昨收
|
||
query = f"""
|
||
SELECT SECCODE, F007N
|
||
FROM ea_trade
|
||
WHERE SECCODE IN ('{codes_str}')
|
||
AND TRADEDATE = (
|
||
SELECT MAX(TRADEDATE) FROM ea_trade WHERE TRADEDATE < '{trade_date}'
|
||
)
|
||
AND F007N IS NOT NULL AND F007N > 0
|
||
"""
|
||
|
||
try:
|
||
with MYSQL_ENGINE.connect() as conn:
|
||
result = conn.execute(text(query))
|
||
return {row[0]: float(row[1]) for row in result if row[1]}
|
||
except Exception as e:
|
||
print(f"获取昨收价失败: {e}")
|
||
return {}
|
||
|
||
|
||
def get_index_prev_close(trade_date: str) -> float:
|
||
"""获取指数昨收价"""
|
||
code_no_suffix = REFERENCE_INDEX.split('.')[0]
|
||
try:
|
||
with MYSQL_ENGINE.connect() as conn:
|
||
result = conn.execute(text("""
|
||
SELECT F006N FROM ea_exchangetrade
|
||
WHERE INDEXCODE = :code AND TRADEDATE < :today
|
||
ORDER BY TRADEDATE DESC LIMIT 1
|
||
"""), {'code': code_no_suffix, 'today': trade_date}).fetchone()
|
||
if result and result[0]:
|
||
return float(result[0])
|
||
except Exception as e:
|
||
print(f"获取指数昨收失败: {e}")
|
||
return None
|
||
|
||
|
||
def get_stock_minute_data(trade_date: str, stock_codes: List[str], since_time: datetime = None) -> pd.DataFrame:
|
||
"""
|
||
从 ClickHouse 获取股票分钟数据
|
||
|
||
Args:
|
||
trade_date: 交易日期
|
||
stock_codes: 股票代码列表
|
||
since_time: 只获取该时间之后的数据（增量获取）
|
||
"""
|
||
client = get_ch_client()
|
||
|
||
ch_codes = []
|
||
code_map = {}
|
||
for code in stock_codes:
|
||
ch_code = code_to_ch_format(code)
|
||
if ch_code:
|
||
ch_codes.append(ch_code)
|
||
code_map[ch_code] = code
|
||
|
||
if not ch_codes:
|
||
return pd.DataFrame()
|
||
|
||
ch_codes_str = "','".join(ch_codes)
|
||
|
||
time_filter = ""
|
||
if since_time:
|
||
time_filter = f"AND timestamp > '{since_time.strftime('%Y-%m-%d %H:%M:%S')}'"
|
||
|
||
query = f"""
|
||
SELECT code, timestamp, close, volume, amt
|
||
FROM stock_minute
|
||
WHERE toDate(timestamp) = '{trade_date}'
|
||
AND code IN ('{ch_codes_str}')
|
||
{time_filter}
|
||
ORDER BY code, timestamp
|
||
"""
|
||
|
||
result = client.execute(query)
|
||
if not result:
|
||
return pd.DataFrame()
|
||
|
||
df = pd.DataFrame(result, columns=['ch_code', 'timestamp', 'close', 'volume', 'amt'])
|
||
df['code'] = df['ch_code'].map(code_map)
|
||
df = df.dropna(subset=['code'])
|
||
return df[['code', 'timestamp', 'close', 'volume', 'amt']]
|
||
|
||
|
||
def get_index_minute_data(trade_date: str, since_time: datetime = None) -> pd.DataFrame:
|
||
"""从 ClickHouse 获取指数分钟数据"""
|
||
client = get_ch_client()
|
||
|
||
time_filter = ""
|
||
if since_time:
|
||
time_filter = f"AND timestamp > '{since_time.strftime('%Y-%m-%d %H:%M:%S')}'"
|
||
|
||
query = f"""
|
||
SELECT timestamp, close, volume, amt
|
||
FROM index_minute
|
||
WHERE toDate(timestamp) = '{trade_date}'
|
||
AND code = '{REFERENCE_INDEX}'
|
||
{time_filter}
|
||
ORDER BY timestamp
|
||
"""
|
||
|
||
result = client.execute(query)
|
||
if not result:
|
||
return pd.DataFrame()
|
||
|
||
return pd.DataFrame(result, columns=['timestamp', 'close', 'volume', 'amt'])
|
||
|
||
|
||
# ==================== 规则评分 ====================
|
||
|
||
def get_size_adjusted_thresholds(stock_count: np.ndarray) -> np.ndarray:
|
||
"""根据概念股票数量计算动态阈值"""
|
||
n = len(stock_count)
|
||
size_factor = np.ones(n)
|
||
|
||
size_factor[stock_count < 5] = 1.8
|
||
size_factor[(stock_count >= 5) & (stock_count < 10)] = 1.4
|
||
size_factor[(stock_count >= 10) & (stock_count < 20)] = 1.2
|
||
size_factor[(stock_count >= 20) & (stock_count < 50)] = 1.0
|
||
size_factor[(stock_count >= 50) & (stock_count < 100)] = 0.85
|
||
size_factor[stock_count >= 100] = 0.7
|
||
|
||
return size_factor
|
||
|
||
|
||
def score_rules_batch(df: pd.DataFrame) -> Tuple[np.ndarray, List[List[str]]]:
|
||
"""批量计算规则得分"""
|
||
n = len(df)
|
||
scores = np.zeros(n)
|
||
triggered = [[] for _ in range(n)]
|
||
|
||
alpha = df['alpha'].values
|
||
alpha_delta = df['alpha_delta'].values
|
||
amt_ratio = df['amt_ratio'].values
|
||
rank_pct = df['rank_pct'].values
|
||
limit_up_ratio = df['limit_up_ratio'].values
|
||
stock_count = df['stock_count'].values if 'stock_count' in df.columns else np.full(n, 20)
|
||
|
||
alpha_abs = np.abs(alpha)
|
||
alpha_delta_abs = np.abs(alpha_delta)
|
||
size_factor = get_size_adjusted_thresholds(stock_count)
|
||
|
||
# Alpha 规则
|
||
alpha_extreme_thresh = 5.0 * size_factor
|
||
mask = alpha_abs >= alpha_extreme_thresh
|
||
scores[mask] += 20
|
||
for i in np.where(mask)[0]: triggered[i].append('alpha_extreme')
|
||
|
||
alpha_strong_thresh = 4.0 * size_factor
|
||
mask = (alpha_abs >= alpha_strong_thresh) & (alpha_abs < alpha_extreme_thresh)
|
||
scores[mask] += 15
|
||
for i in np.where(mask)[0]: triggered[i].append('alpha_strong')
|
||
|
||
alpha_medium_thresh = 3.0 * size_factor
|
||
mask = (alpha_abs >= alpha_medium_thresh) & (alpha_abs < alpha_strong_thresh)
|
||
scores[mask] += 10
|
||
for i in np.where(mask)[0]: triggered[i].append('alpha_medium')
|
||
|
||
# Alpha 加速度
|
||
delta_strong_thresh = 2.0 * size_factor
|
||
mask = alpha_delta_abs >= delta_strong_thresh
|
||
scores[mask] += 15
|
||
for i in np.where(mask)[0]: triggered[i].append('alpha_delta_strong')
|
||
|
||
delta_medium_thresh = 1.5 * size_factor
|
||
mask = (alpha_delta_abs >= delta_medium_thresh) & (alpha_delta_abs < delta_strong_thresh)
|
||
scores[mask] += 10
|
||
for i in np.where(mask)[0]: triggered[i].append('alpha_delta_medium')
|
||
|
||
# 成交额
|
||
mask = amt_ratio >= 10.0
|
||
scores[mask] += 20
|
||
for i in np.where(mask)[0]: triggered[i].append('volume_extreme')
|
||
|
||
mask = (amt_ratio >= 6.0) & (amt_ratio < 10.0)
|
||
scores[mask] += 12
|
||
for i in np.where(mask)[0]: triggered[i].append('volume_strong')
|
||
|
||
# 排名
|
||
mask = rank_pct >= 0.98
|
||
scores[mask] += 15
|
||
for i in np.where(mask)[0]: triggered[i].append('rank_top')
|
||
|
||
mask = rank_pct <= 0.02
|
||
scores[mask] += 15
|
||
for i in np.where(mask)[0]: triggered[i].append('rank_bottom')
|
||
|
||
# 涨停
|
||
limit_high_thresh = 0.30 * size_factor
|
||
mask = limit_up_ratio >= limit_high_thresh
|
||
scores[mask] += 20
|
||
for i in np.where(mask)[0]: triggered[i].append('limit_up_high')
|
||
|
||
limit_medium_thresh = 0.20 * size_factor
|
||
mask = (limit_up_ratio >= limit_medium_thresh) & (limit_up_ratio < limit_high_thresh)
|
||
scores[mask] += 12
|
||
for i in np.where(mask)[0]: triggered[i].append('limit_up_medium')
|
||
|
||
# 概念规模加分
|
||
large_concept = stock_count >= 50
|
||
has_signal = scores > 0
|
||
mask = large_concept & has_signal
|
||
scores[mask] += 10
|
||
for i in np.where(mask)[0]: triggered[i].append('large_concept_bonus')
|
||
|
||
xlarge_concept = stock_count >= 100
|
||
mask = xlarge_concept & has_signal
|
||
scores[mask] += 10
|
||
for i in np.where(mask)[0]: triggered[i].append('xlarge_concept_bonus')
|
||
|
||
# 组合规则
|
||
combo_alpha_thresh = 3.0 * size_factor
|
||
mask = (alpha_abs >= combo_alpha_thresh) & (amt_ratio >= 5.0) & ((rank_pct >= 0.95) | (rank_pct <= 0.05))
|
||
scores[mask] += 20
|
||
for i in np.where(mask)[0]: triggered[i].append('triple_signal')
|
||
|
||
mask = (alpha_abs >= combo_alpha_thresh) & (limit_up_ratio >= 0.15 * size_factor)
|
||
scores[mask] += 15
|
||
for i in np.where(mask)[0]: triggered[i].append('alpha_with_limit')
|
||
|
||
# 小概念惩罚
|
||
tiny_concept = stock_count < 5
|
||
single_rule = np.array([len(t) <= 1 for t in triggered])
|
||
mask = tiny_concept & single_rule & (scores > 0)
|
||
scores[mask] *= 0.5
|
||
for i in np.where(mask)[0]: triggered[i].append('tiny_concept_penalty')
|
||
|
||
scores = np.clip(scores, 0, 100)
|
||
return scores, triggered
|
||
|
||
|
||
def rule_score_with_details(features: Dict, stock_count: int = 50) -> Tuple[float, Dict[str, float]]:
|
||
"""
|
||
单条记录的规则评分（带详情）
|
||
|
||
Args:
|
||
features: 特征字典，包含 alpha, alpha_delta, amt_ratio, rank_pct, limit_up_ratio
|
||
stock_count: 概念股票数量
|
||
|
||
Returns:
|
||
(score, details): 总分和各规则触发详情
|
||
"""
|
||
score = 0.0
|
||
details = {}
|
||
|
||
alpha = features.get('alpha', 0)
|
||
alpha_delta = features.get('alpha_delta', 0)
|
||
amt_ratio = features.get('amt_ratio', 1)
|
||
rank_pct = features.get('rank_pct', 0.5)
|
||
limit_up_ratio = features.get('limit_up_ratio', 0)
|
||
|
||
alpha_abs = abs(alpha)
|
||
alpha_delta_abs = abs(alpha_delta)
|
||
size_factor = get_size_adjusted_thresholds(np.array([stock_count]))[0]
|
||
|
||
# Alpha 规则
|
||
alpha_extreme_thresh = 5.0 * size_factor
|
||
alpha_strong_thresh = 4.0 * size_factor
|
||
alpha_medium_thresh = 3.0 * size_factor
|
||
|
||
if alpha_abs >= alpha_extreme_thresh:
|
||
score += 20
|
||
details['alpha_extreme'] = 20
|
||
elif alpha_abs >= alpha_strong_thresh:
|
||
score += 15
|
||
details['alpha_strong'] = 15
|
||
elif alpha_abs >= alpha_medium_thresh:
|
||
score += 10
|
||
details['alpha_medium'] = 10
|
||
|
||
# Alpha 加速度
|
||
delta_strong_thresh = 2.0 * size_factor
|
||
delta_medium_thresh = 1.5 * size_factor
|
||
|
||
if alpha_delta_abs >= delta_strong_thresh:
|
||
score += 15
|
||
details['alpha_delta_strong'] = 15
|
||
elif alpha_delta_abs >= delta_medium_thresh:
|
||
score += 10
|
||
details['alpha_delta_medium'] = 10
|
||
|
||
# 成交额
|
||
if amt_ratio >= 10.0:
|
||
score += 20
|
||
details['volume_extreme'] = 20
|
||
elif amt_ratio >= 6.0:
|
||
score += 12
|
||
details['volume_strong'] = 12
|
||
|
||
# 排名
|
||
if rank_pct >= 0.98:
|
||
score += 15
|
||
details['rank_top'] = 15
|
||
elif rank_pct <= 0.02:
|
||
score += 15
|
||
details['rank_bottom'] = 15
|
||
|
||
# 涨停
|
||
limit_high_thresh = 0.30 * size_factor
|
||
limit_medium_thresh = 0.20 * size_factor
|
||
|
||
if limit_up_ratio >= limit_high_thresh:
|
||
score += 20
|
||
details['limit_up_high'] = 20
|
||
elif limit_up_ratio >= limit_medium_thresh:
|
||
score += 12
|
||
details['limit_up_medium'] = 12
|
||
|
||
# 概念规模加分
|
||
if score > 0:
|
||
if stock_count >= 50:
|
||
score += 10
|
||
details['large_concept_bonus'] = 10
|
||
if stock_count >= 100:
|
||
score += 10
|
||
details['xlarge_concept_bonus'] = 10
|
||
|
||
# 组合规则
|
||
combo_alpha_thresh = 3.0 * size_factor
|
||
|
||
if alpha_abs >= combo_alpha_thresh and amt_ratio >= 5.0 and (rank_pct >= 0.95 or rank_pct <= 0.05):
|
||
score += 20
|
||
details['triple_signal'] = 20
|
||
|
||
if alpha_abs >= combo_alpha_thresh and limit_up_ratio >= 0.15 * size_factor:
|
||
score += 15
|
||
details['alpha_with_limit'] = 15
|
||
|
||
# 小概念惩罚
|
||
if stock_count < 5 and len(details) <= 1 and score > 0:
|
||
penalty = score * 0.5
|
||
score *= 0.5
|
||
details['tiny_concept_penalty'] = -penalty
|
||
|
||
score = min(max(score, 0), 100)
|
||
return score, details
|
||
|
||
|
||
# ==================== ML 评分器 ====================
|
||
|
||
class MLScorer:
|
||
def __init__(self, checkpoint_dir: str = 'ml/checkpoints', device: str = 'auto'):
|
||
self.checkpoint_dir = Path(checkpoint_dir)
|
||
if device == 'auto':
|
||
self.device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
|
||
else:
|
||
self.device = torch.device(device)
|
||
|
||
self.model = None
|
||
self.thresholds = None
|
||
self._load_model()
|
||
|
||
def _load_model(self):
|
||
model_path = self.checkpoint_dir / 'best_model.pt'
|
||
thresholds_path = self.checkpoint_dir / 'thresholds.json'
|
||
config_path = self.checkpoint_dir / 'config.json'
|
||
|
||
if not model_path.exists():
|
||
print(f"警告: 模型不存在 {model_path}")
|
||
return
|
||
|
||
try:
|
||
from model import LSTMAutoencoder
|
||
|
||
config = {}
|
||
if config_path.exists():
|
||
with open(config_path) as f:
|
||
config = json.load(f).get('model', {})
|
||
|
||
if 'd_model' in config:
|
||
config['hidden_dim'] = config.pop('d_model') // 2
|
||
for key in ['num_encoder_layers', 'num_decoder_layers', 'nhead', 'dim_feedforward', 'max_seq_len', 'use_instance_norm']:
|
||
config.pop(key, None)
|
||
if 'num_layers' not in config:
|
||
config['num_layers'] = 1
|
||
|
||
checkpoint = torch.load(model_path, map_location='cpu')
|
||
self.model = LSTMAutoencoder(**config)
|
||
self.model.load_state_dict(checkpoint['model_state_dict'])
|
||
self.model.to(self.device)
|
||
self.model.eval()
|
||
|
||
if thresholds_path.exists():
|
||
with open(thresholds_path) as f:
|
||
self.thresholds = json.load(f)
|
||
|
||
print(f"ML模型加载成功 (设备: {self.device})")
|
||
except Exception as e:
|
||
print(f"ML模型加载失败: {e}")
|
||
|
||
def is_ready(self):
|
||
return self.model is not None
|
||
|
||
@torch.no_grad()
|
||
def score_batch(self, sequences: np.ndarray, debug: bool = False) -> np.ndarray:
|
||
if not self.is_ready() or len(sequences) == 0:
|
||
return np.zeros(len(sequences))
|
||
|
||
x = torch.FloatTensor(sequences).to(self.device)
|
||
output, _ = self.model(x)
|
||
mse = ((output - x) ** 2).mean(dim=-1)
|
||
errors = mse[:, -1].cpu().numpy()
|
||
|
||
p95 = self.thresholds.get('p95', 0.1) if self.thresholds else 0.1
|
||
scores = np.clip(errors / p95 * 50, 0, 100)
|
||
|
||
if debug and len(errors) > 0:
|
||
print(f"[ML调试] p95={p95:.4f}, errors: min={errors.min():.4f}, max={errors.max():.4f}, mean={errors.mean():.4f}")
|
||
print(f"[ML调试] scores: min={scores.min():.0f}, max={scores.max():.0f}, mean={scores.mean():.0f}, =100占比={100*(scores>=100).mean():.1f}%")
|
||
|
||
return scores
|
||
|
||
|
||
# ==================== 内存数据管理器 ====================
|
||
|
||
class RealtimeDataManager:
|
||
"""
|
||
内存数据管理器
|
||
|
||
- 缓存股票分钟数据和指数数据
|
||
- 增量获取新数据
|
||
- 实时计算概念特征
|
||
"""
|
||
|
||
def __init__(self, concepts: List[dict], prev_close: Dict[str, float], index_prev_close: float):
|
||
self.concepts = concepts
|
||
self.prev_close = prev_close
|
||
self.index_prev_close = index_prev_close
|
||
|
||
# 概念到股票的映射
|
||
self.concept_stocks = {c['concept_id']: set(c['stocks']) for c in concepts}
|
||
self.all_stocks = list(set(s for c in concepts for s in c['stocks']))
|
||
|
||
# 内存缓存：股票分钟数据
|
||
self.stock_data = pd.DataFrame() # code, timestamp, close, volume, amt, change_pct
|
||
self.index_data = pd.DataFrame() # timestamp, close, change_pct
|
||
|
||
# 最后更新时间
|
||
self.last_update_time = None
|
||
|
||
# 概念历史（用于计算变化率）
|
||
self.concept_history = defaultdict(lambda: {'alpha': [], 'amt': []})
|
||
|
||
# 概念特征时间序列（用于 ML）
|
||
self.concept_features_history = defaultdict(list) # concept_id -> list of feature dicts
|
||
|
||
def update(self, trade_date: str) -> int:
|
||
"""
|
||
增量更新数据
|
||
|
||
Returns:
|
||
新增的时间点数量
|
||
"""
|
||
# 获取增量数据
|
||
new_stock_df = get_stock_minute_data(trade_date, self.all_stocks, self.last_update_time)
|
||
new_index_df = get_index_minute_data(trade_date, self.last_update_time)
|
||
|
||
if new_stock_df.empty and new_index_df.empty:
|
||
return 0
|
||
|
||
# 计算涨跌幅
|
||
if not new_stock_df.empty:
|
||
new_stock_df['prev_close'] = new_stock_df['code'].map(self.prev_close)
|
||
new_stock_df = new_stock_df.dropna(subset=['prev_close'])
|
||
new_stock_df['change_pct'] = (new_stock_df['close'] - new_stock_df['prev_close']) / new_stock_df['prev_close'] * 100
|
||
|
||
# 合并到缓存
|
||
self.stock_data = pd.concat([self.stock_data, new_stock_df], ignore_index=True)
|
||
self.stock_data = self.stock_data.drop_duplicates(subset=['code', 'timestamp'], keep='last')
|
||
|
||
if not new_index_df.empty:
|
||
new_index_df['change_pct'] = (new_index_df['close'] - self.index_prev_close) / self.index_prev_close * 100
|
||
|
||
# 调试：打印指数涨跌幅范围
|
||
if len(self.index_data) == 0: # 第一次
|
||
print(f"[调试] 指数 close 范围: {new_index_df['close'].min():.2f} ~ {new_index_df['close'].max():.2f}")
|
||
print(f"[调试] 指数 change_pct 范围: {new_index_df['change_pct'].min():.2f}% ~ {new_index_df['change_pct'].max():.2f}%")
|
||
|
||
self.index_data = pd.concat([self.index_data, new_index_df], ignore_index=True)
|
||
self.index_data = self.index_data.drop_duplicates(subset=['timestamp'], keep='last')
|
||
|
||
# 更新最后时间
|
||
if not new_stock_df.empty:
|
||
self.last_update_time = new_stock_df['timestamp'].max()
|
||
elif not new_index_df.empty:
|
||
self.last_update_time = new_index_df['timestamp'].max()
|
||
|
||
# 获取新时间点
|
||
new_timestamps = sorted(new_stock_df['timestamp'].unique()) if not new_stock_df.empty else []
|
||
|
||
# 计算新时间点的概念特征
|
||
for ts in new_timestamps:
|
||
self._compute_features_for_timestamp(ts)
|
||
|
||
return len(new_timestamps)
|
||
|
||
def _compute_features_for_timestamp(self, ts):
|
||
"""计算单个时间点的概念特征"""
|
||
ts_stock_data = self.stock_data[self.stock_data['timestamp'] == ts]
|
||
index_row = self.index_data[self.index_data['timestamp'] == ts]
|
||
|
||
if ts_stock_data.empty or index_row.empty:
|
||
return
|
||
|
||
index_change = index_row['change_pct'].values[0]
|
||
stock_change = dict(zip(ts_stock_data['code'], ts_stock_data['change_pct']))
|
||
stock_amt = dict(zip(ts_stock_data['code'], ts_stock_data['amt']))
|
||
|
||
for concept_id, stocks in self.concept_stocks.items():
|
||
concept_changes = [stock_change[s] for s in stocks if s in stock_change]
|
||
concept_amts = [stock_amt.get(s, 0) for s in stocks if s in stock_change]
|
||
|
||
if not concept_changes:
|
||
continue
|
||
|
||
avg_change = np.mean(concept_changes)
|
||
total_amt = sum(concept_amts)
|
||
alpha = avg_change - index_change
|
||
|
||
# 涨停比例
|
||
limit_up_count = sum(1 for c in concept_changes if c >= FEATURE_CONFIG['limit_up_threshold'])
|
||
limit_up_ratio = limit_up_count / len(concept_changes)
|
||
|
||
# 更新历史
|
||
history = self.concept_history[concept_id]
|
||
history['alpha'].append(alpha)
|
||
history['amt'].append(total_amt)
|
||
|
||
# 计算变化率
|
||
alpha_delta = 0
|
||
if len(history['alpha']) > FEATURE_CONFIG['alpha_delta_window']:
|
||
alpha_delta = alpha - history['alpha'][-FEATURE_CONFIG['alpha_delta_window'] - 1]
|
||
|
||
amt_ratio = 1.0
|
||
amt_delta = 0
|
||
if len(history['amt']) > FEATURE_CONFIG['amt_ma_window']:
|
||
amt_ma = np.mean(history['amt'][-FEATURE_CONFIG['amt_ma_window'] - 1:-1])
|
||
if amt_ma > 0:
|
||
amt_ratio = total_amt / amt_ma
|
||
amt_delta = total_amt - history['amt'][-2] if len(history['amt']) > 1 else 0
|
||
|
||
features = {
|
||
'timestamp': ts,
|
||
'concept_id': concept_id,
|
||
'alpha': alpha,
|
||
'alpha_delta': alpha_delta,
|
||
'amt_ratio': amt_ratio,
|
||
'amt_delta': amt_delta,
|
||
'limit_up_ratio': limit_up_ratio,
|
||
'stock_count': len(concept_changes),
|
||
'total_amt': total_amt,
|
||
}
|
||
|
||
self.concept_features_history[concept_id].append(features)
|
||
|
||
def get_latest_features(self) -> pd.DataFrame:
|
||
"""获取最新时间点的所有概念特征"""
|
||
if not self.concept_features_history:
|
||
return pd.DataFrame()
|
||
|
||
latest_features = []
|
||
for concept_id, history in self.concept_features_history.items():
|
||
if history:
|
||
latest_features.append(history[-1])
|
||
|
||
if not latest_features:
|
||
return pd.DataFrame()
|
||
|
||
df = pd.DataFrame(latest_features)
|
||
|
||
# 计算排名百分位
|
||
if len(df) > 1:
|
||
df['rank_pct'] = df['alpha'].rank(pct=True)
|
||
else:
|
||
df['rank_pct'] = 0.5
|
||
|
||
return df
|
||
|
||
def get_sequences_for_concepts(self, seq_len: int) -> Tuple[np.ndarray, pd.DataFrame]:
|
||
"""获取所有概念的特征序列（用于 ML 评分）"""
|
||
sequences = []
|
||
infos = []
|
||
|
||
for concept_id, history in self.concept_features_history.items():
|
||
if len(history) < seq_len:
|
||
continue
|
||
|
||
# 取最近 seq_len 个时间点
|
||
recent = history[-seq_len:]
|
||
|
||
# 构建序列
|
||
seq = np.array([[
|
||
f['alpha'],
|
||
f['alpha_delta'],
|
||
f['amt_ratio'],
|
||
f['amt_delta'],
|
||
f.get('rank_pct', 0.5),
|
||
f['limit_up_ratio']
|
||
] for f in recent])
|
||
|
||
seq = np.nan_to_num(seq, nan=0.0, posinf=0.0, neginf=0.0)
|
||
seq = np.clip(seq, -CONFIG['clip_value'], CONFIG['clip_value'])
|
||
|
||
sequences.append(seq)
|
||
infos.append(recent[-1]) # 最新特征
|
||
|
||
if not sequences:
|
||
return np.array([]), pd.DataFrame()
|
||
|
||
# 补充 rank_pct
|
||
info_df = pd.DataFrame(infos)
|
||
if 'rank_pct' not in info_df.columns and len(info_df) > 1:
|
||
info_df['rank_pct'] = info_df['alpha'].rank(pct=True)
|
||
|
||
return np.array(sequences), info_df
|
||
|
||
def get_all_timestamps(self) -> List:
|
||
"""获取所有时间点"""
|
||
if self.stock_data.empty:
|
||
return []
|
||
return sorted(self.stock_data['timestamp'].unique())
|
||
|
||
def get_concept_features_df(self) -> pd.DataFrame:
|
||
"""获取概念特征的 DataFrame 形式（用于批量回测）"""
|
||
if not self.concept_features_history:
|
||
return pd.DataFrame()
|
||
|
||
rows = []
|
||
for concept_id, history in self.concept_features_history.items():
|
||
for f in history:
|
||
row = {
|
||
'concept_id': concept_id,
|
||
'timestamp': f['timestamp'],
|
||
'alpha': f['alpha'],
|
||
'alpha_delta': f['alpha_delta'],
|
||
'amt_ratio': f['amt_ratio'],
|
||
'amt_delta': f.get('amt_delta', 0),
|
||
'limit_up_ratio': f['limit_up_ratio'],
|
||
'stock_count': f.get('stock_count', 0),
|
||
'total_amt': f.get('total_amt', 0),
|
||
}
|
||
rows.append(row)
|
||
|
||
if not rows:
|
||
return pd.DataFrame()
|
||
|
||
df = pd.DataFrame(rows)
|
||
|
||
# 按时间点计算 rank_pct（每个时间点内部排名）
|
||
df['rank_pct'] = df.groupby('timestamp')['alpha'].rank(pct=True)
|
||
|
||
return df
|
||
|
||
|
||
# ==================== 冷却期管理 ====================
|
||
|
||
class CooldownManager:
|
||
def __init__(self, cooldown_minutes: int = 8):
|
||
self.cooldown_minutes = cooldown_minutes
|
||
self.last_alert_time = {}
|
||
|
||
def is_in_cooldown(self, concept_id: str, current_time: datetime) -> bool:
|
||
if concept_id not in self.last_alert_time:
|
||
return False
|
||
last_time = self.last_alert_time[concept_id]
|
||
diff = (current_time - last_time).total_seconds() / 60
|
||
return diff < self.cooldown_minutes
|
||
|
||
def record_alert(self, concept_id: str, alert_time: datetime):
|
||
self.last_alert_time[concept_id] = alert_time
|
||
|
||
def cleanup_old(self, current_time: datetime):
|
||
cutoff = current_time - timedelta(minutes=self.cooldown_minutes * 2)
|
||
self.last_alert_time = {cid: t for cid, t in self.last_alert_time.items() if t > cutoff}
|
||
|
||
|
||
# ==================== 异动检测 ====================
|
||
|
||
def detect_anomalies(
|
||
ml_scorer: MLScorer,
|
||
data_mgr: RealtimeDataManager,
|
||
cooldown_mgr: CooldownManager,
|
||
trade_date: str,
|
||
config: Dict
|
||
) -> List[Dict]:
|
||
"""检测当前时刻的异动"""
|
||
|
||
# 获取最新特征
|
||
latest_df = data_mgr.get_latest_features()
|
||
if latest_df.empty:
|
||
return []
|
||
|
||
# 获取 ML 序列
|
||
sequences, info_df = data_mgr.get_sequences_for_concepts(config['seq_len'])
|
||
|
||
if len(sequences) == 0:
|
||
return []
|
||
|
||
# 获取当前时间
|
||
current_time = pd.to_datetime(info_df['timestamp'].iloc[0])
|
||
|
||
# 清理过期冷却
|
||
cooldown_mgr.cleanup_old(current_time)
|
||
|
||
# 过滤冷却中的概念
|
||
valid_mask = []
|
||
for _, row in info_df.iterrows():
|
||
in_cooldown = cooldown_mgr.is_in_cooldown(row['concept_id'], current_time)
|
||
valid_mask.append(not in_cooldown)
|
||
|
||
valid_mask = np.array(valid_mask)
|
||
sequences = sequences[valid_mask]
|
||
info_df = info_df[valid_mask].reset_index(drop=True)
|
||
|
||
if len(sequences) == 0:
|
||
return []
|
||
|
||
# 过滤小波动
|
||
alpha_mask = np.abs(info_df['alpha'].values) >= config['min_alpha_abs']
|
||
sequences = sequences[alpha_mask]
|
||
info_df = info_df[alpha_mask].reset_index(drop=True)
|
||
|
||
if len(sequences) == 0:
|
||
return []
|
||
|
||
# 规则评分
|
||
rule_scores, triggered_rules = score_rules_batch(info_df)
|
||
|
||
# ML 评分
|
||
ml_scores = ml_scorer.score_batch(sequences)
|
||
|
||
# 融合得分
|
||
w1, w2 = config['rule_weight'], config['ml_weight']
|
||
final_scores = w1 * rule_scores + w2 * ml_scores
|
||
|
||
# 判断异动
|
||
alerts = []
|
||
for i, row in info_df.iterrows():
|
||
rule_score = rule_scores[i]
|
||
ml_score = ml_scores[i]
|
||
final_score = final_scores[i]
|
||
|
||
is_anomaly = (
|
||
rule_score >= config['rule_trigger'] or
|
||
ml_score >= config['ml_trigger'] or
|
||
final_score >= config['fusion_trigger']
|
||
)
|
||
|
||
if not is_anomaly:
|
||
continue
|
||
|
||
# 触发原因
|
||
if rule_score >= config['rule_trigger']:
|
||
trigger = f'规则强信号({rule_score:.0f}分)'
|
||
elif ml_score >= config['ml_trigger']:
|
||
trigger = f'ML强信号({ml_score:.0f}分)'
|
||
else:
|
||
trigger = f'融合触发({final_score:.0f}分)'
|
||
|
||
# 异动类型
|
||
alpha = row['alpha']
|
||
if alpha >= 1.5:
|
||
alert_type = 'surge_up'
|
||
elif alpha <= -1.5:
|
||
alert_type = 'surge_down'
|
||
elif row['amt_ratio'] >= 3.0:
|
||
alert_type = 'volume_spike'
|
||
else:
|
||
alert_type = 'unknown'
|
||
|
||
alert = {
|
||
'concept_id': row['concept_id'],
|
||
'alert_time': row['timestamp'],
|
||
'trade_date': trade_date,
|
||
'alert_type': alert_type,
|
||
'final_score': final_score,
|
||
'rule_score': rule_score,
|
||
'ml_score': ml_score,
|
||
'trigger_reason': trigger,
|
||
'triggered_rules': triggered_rules[i],
|
||
'alpha': row['alpha'],
|
||
'alpha_delta': row['alpha_delta'],
|
||
'amt_ratio': row['amt_ratio'],
|
||
'amt_delta': row.get('amt_delta', 0),
|
||
'rank_pct': row.get('rank_pct', 0.5),
|
||
'limit_up_ratio': row['limit_up_ratio'],
|
||
'stock_count': row['stock_count'],
|
||
'total_amt': row['total_amt'],
|
||
}
|
||
|
||
alerts.append(alert)
|
||
cooldown_mgr.record_alert(row['concept_id'], current_time)
|
||
|
||
# 按得分排序
|
||
alerts.sort(key=lambda x: x['final_score'], reverse=True)
|
||
return alerts[:config['max_alerts_per_minute']]
|
||
|
||
|
||
# ==================== 数据库写入 ====================
|
||
|
||
def save_alerts_to_mysql(alerts: List[Dict]) -> int:
|
||
if not alerts:
|
||
return 0
|
||
|
||
saved = 0
|
||
with MYSQL_ENGINE.begin() as conn:
|
||
for alert in alerts:
|
||
try:
|
||
insert_sql = text("""
|
||
INSERT IGNORE INTO concept_anomaly_hybrid
|
||
(concept_id, alert_time, trade_date, alert_type,
|
||
final_score, rule_score, ml_score, trigger_reason,
|
||
alpha, alpha_delta, amt_ratio, amt_delta,
|
||
rank_pct, limit_up_ratio, stock_count, total_amt,
|
||
triggered_rules)
|
||
VALUES
|
||
(:concept_id, :alert_time, :trade_date, :alert_type,
|
||
:final_score, :rule_score, :ml_score, :trigger_reason,
|
||
:alpha, :alpha_delta, :amt_ratio, :amt_delta,
|
||
:rank_pct, :limit_up_ratio, :stock_count, :total_amt,
|
||
:triggered_rules)
|
||
""")
|
||
|
||
result = conn.execute(insert_sql, {
|
||
'concept_id': alert['concept_id'],
|
||
'alert_time': alert['alert_time'],
|
||
'trade_date': alert['trade_date'],
|
||
'alert_type': alert['alert_type'],
|
||
'final_score': alert['final_score'],
|
||
'rule_score': alert['rule_score'],
|
||
'ml_score': alert['ml_score'],
|
||
'trigger_reason': alert['trigger_reason'],
|
||
'alpha': alert.get('alpha', 0),
|
||
'alpha_delta': alert.get('alpha_delta', 0),
|
||
'amt_ratio': alert.get('amt_ratio', 1),
|
||
'amt_delta': alert.get('amt_delta', 0),
|
||
'rank_pct': alert.get('rank_pct', 0.5),
|
||
'limit_up_ratio': alert.get('limit_up_ratio', 0),
|
||
'stock_count': alert.get('stock_count', 0),
|
||
'total_amt': alert.get('total_amt', 0),
|
||
'triggered_rules': json.dumps(alert.get('triggered_rules', []), ensure_ascii=False),
|
||
})
|
||
|
||
if result.rowcount > 0:
|
||
saved += 1
|
||
except Exception as e:
|
||
print(f"保存失败: {alert['concept_id']} - {e}")
|
||
|
||
return saved
|
||
|
||
|
||
# ==================== 主服务 ====================
|
||
|
||
class RealtimeDetectorService:
|
||
def __init__(self, checkpoint_dir: str = 'ml/checkpoints', device: str = 'auto'):
|
||
self.checkpoint_dir = checkpoint_dir
|
||
self.device = device
|
||
|
||
# 初始化 ML 评分器
|
||
self.ml_scorer = MLScorer(checkpoint_dir, device)
|
||
|
||
# 这些在 init_for_trade_date 中初始化
|
||
self.data_mgr = None
|
||
self.cooldown_mgr = None
|
||
self.trade_date = None
|
||
|
||
def init_for_trade_date(self, trade_date: str, preload_history: bool = True):
|
||
"""
|
||
为指定交易日初始化
|
||
|
||
Args:
|
||
trade_date: 交易日期
|
||
preload_history: 是否预加载当天已有的历史数据（实盘必须为 True）
|
||
"""
|
||
if self.trade_date == trade_date and self.data_mgr is not None:
|
||
return
|
||
|
||
print(f"[初始化] 交易日: {trade_date}")
|
||
|
||
# 获取概念列表
|
||
print(f"[初始化] 获取概念列表...")
|
||
concepts = get_all_concepts()
|
||
|
||
# 获取所有股票
|
||
all_stocks = list(set(s for c in concepts for s in c['stocks']))
|
||
print(f"[初始化] 共 {len(all_stocks)} 只股票")
|
||
|
||
# 获取昨收价
|
||
print(f"[初始化] 获取昨收价...")
|
||
prev_close = get_prev_close(all_stocks, trade_date)
|
||
index_prev_close = get_index_prev_close(trade_date)
|
||
print(f"[初始化] 获取到 {len(prev_close)} 只股票的昨收价")
|
||
print(f"[初始化] 指数昨收价: {index_prev_close}")
|
||
|
||
# 创建数据管理器
|
||
self.data_mgr = RealtimeDataManager(concepts, prev_close, index_prev_close)
|
||
self.cooldown_mgr = CooldownManager(CONFIG['cooldown_minutes'])
|
||
self.trade_date = trade_date
|
||
|
||
# 预加载当天已有的历史数据（实盘关键）
|
||
if preload_history:
|
||
self._preload_today_history(trade_date)
|
||
|
||
def _preload_today_history(self, trade_date: str):
|
||
"""
|
||
预加载当天已有的历史数据到内存
|
||
|
||
这是实盘运行的关键：
|
||
- 在盘中任意时刻启动服务时，需要先加载当天已有的数据
|
||
- 这样才能正确计算 alpha_delta（需要过去 5 分钟）和 amt_ratio（需要过去 20 分钟）
|
||
- 以及构建 ML 所需的序列（需要 seq_len=15 分钟）
|
||
|
||
整个过程不依赖 prepare_data.py，直接从 ClickHouse 读取原始数据计算
|
||
"""
|
||
print(f"[预热] 加载当天历史数据...")
|
||
|
||
# 直接调用 update，但不设置 last_update_time，会获取当天所有数据
|
||
# data_mgr.last_update_time 初始为 None，会获取全部数据
|
||
n_updates = self.data_mgr.update(trade_date)
|
||
|
||
if n_updates > 0:
|
||
print(f"[预热] 加载完成，共 {n_updates} 个时间点")
|
||
|
||
# 检查是否满足 ML 所需的最小数据量
|
||
min_required = CONFIG['seq_len'] + FEATURE_CONFIG['amt_ma_window']
|
||
if n_updates < min_required:
|
||
print(f"[预热] 警告：数据量 {n_updates} < 最小需求 {min_required}，部分特征可能不准确")
|
||
else:
|
||
print(f"[预热] 数据充足，可以正常检测")
|
||
else:
|
||
print(f"[预热] 当天暂无历史数据（可能是开盘前）")
|
||
|
||
def backfill_today(self):
|
||
"""
|
||
补齐当天历史数据并检测异动（回补模式）
|
||
|
||
使用与 backtest_fast.py 完全相同的逻辑：
|
||
1. 先用 prepare_data.py 生成当天的 parquet 文件
|
||
2. 读取 parquet 文件进行回测
|
||
|
||
注意：这个方法用于回补历史异动记录，不是实盘必须的
|
||
实盘模式下，init_for_trade_date 会自动预热历史数据
|
||
"""
|
||
trade_date = get_current_trade_date()
|
||
print(f"[补齐] 交易日: {trade_date}")
|
||
|
||
# 1. 生成当天的 parquet 文件
|
||
parquet_path = Path('ml/data') / f'features_{trade_date}.parquet'
|
||
|
||
if not parquet_path.exists():
|
||
print(f"[补齐] 生成当天特征数据...")
|
||
self._generate_today_parquet(trade_date)
|
||
|
||
if not parquet_path.exists():
|
||
print(f"[补齐] 无法生成特征数据，跳过")
|
||
return
|
||
|
||
# 2. 读取 parquet 文件
|
||
df = pd.read_parquet(parquet_path)
|
||
print(f"[补齐] 读取到 {len(df)} 条特征数据")
|
||
|
||
if df.empty:
|
||
print("[补齐] 无数据")
|
||
return
|
||
|
||
# 打印特征分布（调试）
|
||
print(f"[调试] alpha 分布: min={df['alpha'].min():.2f}, max={df['alpha'].max():.2f}, mean={df['alpha'].mean():.2f}")
|
||
print(f"[调试] |alpha| >= 0.3 的数量: {(df['alpha'].abs() >= 0.3).sum()}")
|
||
|
||
# 3. 使用 backtest_fast.py 相同的回测逻辑
|
||
alerts = self._backtest_from_parquet(df, trade_date)
|
||
|
||
# 4. 保存结果
|
||
if alerts:
|
||
saved = save_alerts_to_mysql(alerts)
|
||
print(f"[补齐] 完成！共 {len(alerts)} 个异动, 保存 {saved} 条")
|
||
|
||
# 统计触发来源
|
||
trigger_stats = {'规则': 0, 'ML': 0, '融合': 0}
|
||
for a in alerts:
|
||
reason = a['trigger_reason']
|
||
if '规则' in reason:
|
||
trigger_stats['规则'] += 1
|
||
elif 'ML' in reason:
|
||
trigger_stats['ML'] += 1
|
||
else:
|
||
trigger_stats['融合'] += 1
|
||
print(f"[补齐] 触发来源: {trigger_stats}")
|
||
else:
|
||
print("[补齐] 无异动")
|
||
|
||
def _generate_today_parquet(self, trade_date: str):
|
||
"""
|
||
生成当天的 parquet 文件（调用 prepare_data.py 的逻辑）
|
||
"""
|
||
import subprocess
|
||
cmd = ['python', 'ml/prepare_data.py', '--start', trade_date, '--end', trade_date]
|
||
print(f"[补齐] 执行: {' '.join(cmd)}")
|
||
try:
|
||
result = subprocess.run(cmd, capture_output=True, text=True, timeout=300)
|
||
if result.returncode != 0:
|
||
print(f"[补齐] prepare_data.py 执行失败: {result.stderr}")
|
||
except Exception as e:
|
||
print(f"[补齐] prepare_data.py 执行异常: {e}")
|
||
|
||
def _backtest_from_parquet(self, df: pd.DataFrame, trade_date: str) -> List[Dict]:
|
||
"""
|
||
从 parquet 数据回测（与 backtest_fast.py 完全一致的逻辑）
|
||
"""
|
||
seq_len = CONFIG['seq_len']
|
||
now = datetime.now()
|
||
|
||
# 确保按概念和时间排序
|
||
df = df.sort_values(['concept_id', 'timestamp']).reset_index(drop=True)
|
||
|
||
# 获取所有时间点
|
||
all_timestamps = sorted(df['timestamp'].unique())
|
||
|
||
# 只处理当前时间之前的
|
||
past_timestamps = []
|
||
for ts in all_timestamps:
|
||
try:
|
||
ts_dt = pd.to_datetime(ts)
|
||
if ts_dt.tzinfo is not None:
|
||
ts_dt = ts_dt.tz_localize(None)
|
||
if ts_dt < now:
|
||
past_timestamps.append(ts)
|
||
except Exception:
|
||
continue
|
||
|
||
print(f"[补齐] 处理 {len(past_timestamps)} 个历史时间点...")
|
||
|
||
if len(past_timestamps) < seq_len:
|
||
print(f"[补齐] 时间点不足 {seq_len}，跳过")
|
||
return []
|
||
|
||
# 构建序列（与 backtest_fast.py 的 build_sequences_fast 一致）
|
||
sequences = []
|
||
infos = []
|
||
|
||
groups = df.groupby('concept_id')
|
||
|
||
for concept_id, gdf in groups:
|
||
gdf = gdf.reset_index(drop=True)
|
||
feat_matrix = gdf[FEATURES].values
|
||
feat_matrix = np.nan_to_num(feat_matrix, nan=0.0, posinf=0.0, neginf=0.0)
|
||
feat_matrix = np.clip(feat_matrix, -CONFIG['clip_value'], CONFIG['clip_value'])
|
||
|
||
n_total = len(feat_matrix)
|
||
if n_total < seq_len:
|
||
continue
|
||
|
||
for i in range(n_total - seq_len + 1):
|
||
seq = feat_matrix[i:i + seq_len]
|
||
row = gdf.iloc[i + seq_len - 1]
|
||
|
||
# 只保留当前时间之前的
|
||
ts = row['timestamp']
|
||
try:
|
||
ts_dt = pd.to_datetime(ts)
|
||
if ts_dt.tzinfo is not None:
|
||
ts_dt = ts_dt.tz_localize(None)
|
||
if ts_dt >= now:
|
||
continue
|
||
except Exception:
|
||
continue
|
||
|
||
sequences.append(seq)
|
||
infos.append({
|
||
'concept_id': concept_id,
|
||
'timestamp': row['timestamp'],
|
||
'alpha': row['alpha'],
|
||
'alpha_delta': row.get('alpha_delta', 0),
|
||
'amt_ratio': row.get('amt_ratio', 1),
|
||
'amt_delta': row.get('amt_delta', 0),
|
||
'rank_pct': row.get('rank_pct', 0.5),
|
||
'limit_up_ratio': row.get('limit_up_ratio', 0),
|
||
'stock_count': row.get('stock_count', 0),
|
||
'total_amt': row.get('total_amt', 0),
|
||
})
|
||
|
||
if not sequences:
|
||
return []
|
||
|
||
sequences = np.array(sequences)
|
||
info_df = pd.DataFrame(infos)
|
||
|
||
print(f"[补齐] 构建了 {len(sequences)} 个序列")
|
||
|
||
# 过滤小波动
|
||
alpha_abs = np.abs(info_df['alpha'].values)
|
||
valid_mask = alpha_abs >= CONFIG['min_alpha_abs']
|
||
sequences = sequences[valid_mask]
|
||
info_df = info_df[valid_mask].reset_index(drop=True)
|
||
|
||
if len(sequences) == 0:
|
||
return []
|
||
|
||
print(f"[补齐] 过滤后 {len(sequences)} 个序列")
|
||
|
||
# 批量规则评分
|
||
rule_scores, triggered_rules = score_rules_batch(info_df)
|
||
|
||
# 批量 ML 评分
|
||
batch_size = 2048
|
||
ml_scores = []
|
||
for i in range(0, len(sequences), batch_size):
|
||
batch_seq = sequences[i:i+batch_size]
|
||
batch_scores = self.ml_scorer.score_batch(batch_seq)
|
||
ml_scores.append(batch_scores)
|
||
ml_scores = np.concatenate(ml_scores) if ml_scores else np.zeros(len(sequences))
|
||
|
||
# 融合得分
|
||
w1, w2 = CONFIG['rule_weight'], CONFIG['ml_weight']
|
||
final_scores = w1 * rule_scores + w2 * ml_scores
|
||
|
||
# 判断异动
|
||
is_anomaly = (
|
||
(rule_scores >= CONFIG['rule_trigger']) |
|
||
(ml_scores >= CONFIG['ml_trigger']) |
|
||
(final_scores >= CONFIG['fusion_trigger'])
|
||
)
|
||
|
||
# 添加分数到 info_df
|
||
info_df['rule_score'] = rule_scores
|
||
info_df['ml_score'] = ml_scores
|
||
info_df['final_score'] = final_scores
|
||
info_df['is_anomaly'] = is_anomaly
|
||
info_df['triggered_rules'] = triggered_rules
|
||
|
||
# 只保留异动
|
||
anomaly_df = info_df[info_df['is_anomaly']].copy()
|
||
|
||
if len(anomaly_df) == 0:
|
||
return []
|
||
|
||
print(f"[补齐] 发现 {len(anomaly_df)} 个候选异动")
|
||
|
||
# 应用冷却期
|
||
anomaly_df = anomaly_df.sort_values(['concept_id', 'timestamp'])
|
||
cooldown = {}
|
||
keep_mask = []
|
||
|
||
for _, row in anomaly_df.iterrows():
|
||
cid = row['concept_id']
|
||
ts = row['timestamp']
|
||
|
||
if cid in cooldown:
|
||
try:
|
||
diff = (pd.to_datetime(ts) - pd.to_datetime(cooldown[cid])).total_seconds() / 60
|
||
except:
|
||
diff = CONFIG['cooldown_minutes'] + 1
|
||
|
||
if diff < CONFIG['cooldown_minutes']:
|
||
keep_mask.append(False)
|
||
continue
|
||
|
||
cooldown[cid] = ts
|
||
keep_mask.append(True)
|
||
|
||
anomaly_df = anomaly_df[keep_mask]
|
||
|
||
print(f"[补齐] 冷却后 {len(anomaly_df)} 个异动")
|
||
|
||
# 按时间分组，每分钟最多 max_alerts_per_minute 个
|
||
alerts = []
|
||
for ts, group in anomaly_df.groupby('timestamp'):
|
||
group = group.nlargest(CONFIG['max_alerts_per_minute'], 'final_score')
|
||
|
||
for _, row in group.iterrows():
|
||
alpha = row['alpha']
|
||
if alpha >= 1.5:
|
||
atype = 'surge_up'
|
||
elif alpha <= -1.5:
|
||
atype = 'surge_down'
|
||
elif row['amt_ratio'] >= 3.0:
|
||
atype = 'volume_spike'
|
||
else:
|
||
atype = 'unknown'
|
||
|
||
rule_score = row['rule_score']
|
||
ml_score = row['ml_score']
|
||
final_score = row['final_score']
|
||
|
||
if rule_score >= CONFIG['rule_trigger']:
|
||
trigger = f'规则强信号({rule_score:.0f}分)'
|
||
elif ml_score >= CONFIG['ml_trigger']:
|
||
trigger = f'ML强信号({ml_score:.0f}分)'
|
||
else:
|
||
trigger = f'融合触发({final_score:.0f}分)'
|
||
|
||
alerts.append({
|
||
'concept_id': row['concept_id'],
|
||
'alert_time': row['timestamp'],
|
||
'trade_date': trade_date,
|
||
'alert_type': atype,
|
||
'final_score': final_score,
|
||
'rule_score': rule_score,
|
||
'ml_score': ml_score,
|
||
'trigger_reason': trigger,
|
||
'triggered_rules': row['triggered_rules'],
|
||
'alpha': alpha,
|
||
'alpha_delta': row['alpha_delta'],
|
||
'amt_ratio': row['amt_ratio'],
|
||
'amt_delta': row['amt_delta'],
|
||
'rank_pct': row['rank_pct'],
|
||
'limit_up_ratio': row['limit_up_ratio'],
|
||
'stock_count': row['stock_count'],
|
||
'total_amt': row['total_amt'],
|
||
})
|
||
|
||
return alerts
|
||
|
||
def run_once(self):
|
||
"""执行一次检测"""
|
||
now = datetime.now()
|
||
trade_date = get_current_trade_date()
|
||
|
||
if not is_trading_time():
|
||
print(f"[{now.strftime('%H:%M:%S')}] 非交易时间，跳过")
|
||
return
|
||
|
||
# 初始化
|
||
self.init_for_trade_date(trade_date)
|
||
|
||
print(f"[{now.strftime('%H:%M:%S')}] 获取新数据...")
|
||
|
||
# 增量更新
|
||
n_updates = self.data_mgr.update(trade_date)
|
||
print(f" 新增 {n_updates} 个时间点")
|
||
|
||
if n_updates == 0:
|
||
print(f" 无新数据")
|
||
return
|
||
|
||
# 检测
|
||
alerts = detect_anomalies(
|
||
self.ml_scorer,
|
||
self.data_mgr,
|
||
self.cooldown_mgr,
|
||
trade_date,
|
||
CONFIG
|
||
)
|
||
|
||
if alerts:
|
||
saved = save_alerts_to_mysql(alerts)
|
||
print(f" 检测到 {len(alerts)} 个异动, 保存 {saved} 条")
|
||
|
||
for alert in alerts[:5]:
|
||
print(f" - {alert['concept_id']}: {alert['alert_type']} "
|
||
f"(final={alert['final_score']:.0f}, rule={alert['rule_score']:.0f}, ml={alert['ml_score']:.0f})")
|
||
else:
|
||
print(f" 无异动")
|
||
|
||
def run_loop(self, backfill: bool = False):
|
||
"""
|
||
持续运行（实盘模式）
|
||
|
||
Args:
|
||
backfill: 是否回补历史异动到数据库（使用 prepare_data.py 方式）
|
||
默认 False，因为实盘模式下 init_for_trade_date 会自动预热数据
|
||
"""
|
||
print("=" * 60)
|
||
print("实时概念异动检测服务（实盘模式）")
|
||
print("=" * 60)
|
||
print(f"模型目录: {self.checkpoint_dir}")
|
||
print(f"交易时段: {TRADING_PERIODS}")
|
||
print(f"ML 序列长度: {CONFIG['seq_len']} 分钟")
|
||
print(f"成交额均值窗口: {FEATURE_CONFIG['amt_ma_window']} 分钟")
|
||
print("=" * 60)
|
||
|
||
# 立即初始化并预热（即使不在交易时间也预热，方便测试）
|
||
trade_date = get_current_trade_date()
|
||
print(f"\n[启动] 初始化交易日 {trade_date}...")
|
||
self.init_for_trade_date(trade_date, preload_history=True)
|
||
|
||
# 可选：回补历史异动记录到数据库
|
||
if backfill and is_trading_time():
|
||
print("\n[启动] 回补历史异动...")
|
||
self.backfill_today()
|
||
|
||
# 每分钟第 10 秒执行
|
||
schedule.every().minute.at(":10").do(self.run_once)
|
||
|
||
print("\n服务已启动，等待下一分钟...")
|
||
|
||
while True:
|
||
schedule.run_pending()
|
||
time.sleep(1)
|
||
|
||
|
||
# ==================== 主函数 ====================
|
||
|
||
def main():
|
||
parser = argparse.ArgumentParser(description='实时概念异动检测')
|
||
parser.add_argument('--checkpoint_dir', default='ml/checkpoints', help='模型目录')
|
||
parser.add_argument('--device', default='auto', help='设备 (auto/cpu/cuda)')
|
||
parser.add_argument('--once', action='store_true', help='只运行一次检测')
|
||
parser.add_argument('--backfill', action='store_true', help='启动时回补历史异动到数据库')
|
||
parser.add_argument('--backfill-only', action='store_true', help='只回补历史（不持续运行）')
|
||
|
||
args = parser.parse_args()
|
||
|
||
service = RealtimeDetectorService(
|
||
checkpoint_dir=args.checkpoint_dir,
|
||
device=args.device
|
||
)
|
||
|
||
if args.once:
|
||
# 单次检测模式
|
||
service.run_once()
|
||
elif args.backfill_only:
|
||
# 仅回补历史模式（需要 prepare_data.py）
|
||
service.backfill_today()
|
||
else:
|
||
# 实盘持续运行模式（自动预热，不依赖 prepare_data.py）
|
||
service.run_loop(backfill=args.backfill)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|