ZLData 量化分析模块 — 详细设计文档

EMA12 = EMA(close, 12)
EMA26 = EMA(close, 26)
DIF = EMA12 - EMA26
DEA = EMA(DIF, 9)
MACD柱 = 2 × (DIF - DEA)

RS = 周期内平均涨幅 / 周期内平均跌幅
RSI = 100 - 100 / (1 + RS)

中轨 = MA(close, 20)
上轨 = 中轨 + 2 × std(close, 20)
下轨 = 中轨 - 2 × std(close, 20)

import pandas_ta as ta
import pandas as pd
 
def compute_all_indicators(df: pd.DataFrame) -> pd.DataFrame:
    """
    对单只股票的日K线 DataFrame 计算全部技术指标。
    df 必须包含: date, open, high, low, close, volume
    返回添加了全部指标列的 DataFrame
    """
    df = df.copy()
    
    # 趋势类
    df['ma5'] = ta.sma(df['close'], length=5)
    df['ma10'] = ta.sma(df['close'], length=10)
    df['ma20'] = ta.sma(df['close'], length=20)
    df['ma60'] = ta.sma(df['close'], length=60)
    df['ma120'] = ta.sma(df['close'], length=120)
    
    # MACD
    macd = ta.macd(df['close'], fast=12, slow=26, signal=9)
    df = pd.concat([df, macd], axis=1)
    
    # RSI
    df['rsi6'] = ta.rsi(df['close'], length=6)
    df['rsi14'] = ta.rsi(df['close'], length=14)
    df['rsi24'] = ta.rsi(df['close'], length=24)
    
    # KDJ
    kdj = ta.kdj(df['high'], df['low'], df['close'], length=9, signal=3)
    df = pd.concat([df, kdj], axis=1)
    
    # 布林带
    bbands = ta.bbands(df['close'], length=20, std=2)
    df = pd.concat([df, bbands], axis=1)
    
    # ATR
    df['atr14'] = ta.atr(df['high'], df['low'], df['close'], length=14)
    
    # OBV
    df['obv'] = ta.obv(df['close'], df['volume'])
    
    # 均线多头排列
    df['ma_bullish'] = (
        (df['ma5'] > df['ma10']) & 
        (df['ma10'] > df['ma20']) & 
        (df['ma20'] > df['ma60'])
    ).astype(int)
    
    return df

def update_indicators_incremental(stock_code: str, since_date: str):
    """
    只更新 since_date 之后的技术指标。
    为实现计算精确，需要回取 N 个历史交易日的数据（N = max_lookback）。
    """
    MAX_LOOKBACK = 120  # MA120 需要 120 个交易日
    
    # 1. 从 DB 取 since_date - MAX_LOOKBACK 以来的全部行情
    df = fetch_daily_kline(stock_code, from_date=offset_date(since_date, -MAX_LOOKBACK))
    
    # 2. 全量计算指标（只写入 since_date 之后的记录）
    df = compute_all_indicators(df)
    df_to_write = df[df['trade_date'] >= since_date]
    
    # 3. 幂等写入指标表
    upsert_indicators(stock_code, df_to_write)

CREATE TABLE trade_stock_indicator_daily (
    id BIGINT IDENTITY(1,1) PRIMARY KEY,
    stock_code VARCHAR(20) NOT NULL,
    trade_date DATE NOT NULL,
    
    -- 趋势类
    ma5 DECIMAL(12,4),
    ma10 DECIMAL(12,4),
    ma20 DECIMAL(12,4),
    ma60 DECIMAL(12,4),
    ma120 DECIMAL(12,4),
    
    -- MACD 系列
    macd_dif DECIMAL(12,6),
    macd_dea DECIMAL(12,6),
    macd_hist DECIMAL(12,6),
    
    -- RSI 系列
    rsi6 DECIMAL(8,4),
    rsi14 DECIMAL(8,4),
    rsi24 DECIMAL(8,4),
    
    -- KDJ 系列
    kdj_k DECIMAL(8,4),
    kdj_d DECIMAL(8,4),
    kdj_j DECIMAL(8,4),
    
    -- 布林带
    boll_upper DECIMAL(12,4),
    boll_middle DECIMAL(12,4),
    boll_lower DECIMAL(12,4),
    
    -- 波动/成交量
    atr14 DECIMAL(12,4),
    obv DECIMAL(20,2),
    ma_bullish TINYINT DEFAULT 0,
    
    -- 元数据
    created_at DATETIME2 DEFAULT GETDATE(),
    updated_at DATETIME2 DEFAULT GETDATE(),
    
    CONSTRAINT uq_stock_indicator_date UNIQUE (stock_code, trade_date)
);
 
CREATE INDEX idx_indicator_date ON trade_stock_indicator_daily(trade_date);
CREATE INDEX idx_indicator_stock ON trade_stock_indicator_daily(stock_code);

def momentum_20d(close_series: pd.Series) -> float:
    """20日动量：今日收盘 / 20日前收盘 - 1"""
    if len(close_series) < 21:
        return None
    return close_series.iloc[-1] / close_series.iloc[-21] - 1
 
def momentum_60d(close_series: pd.Series) -> float:
    """60日动量"""
    if len(close_series) < 61:
        return None
    return close_series.iloc[-1] / close_series.iloc[-61] - 1

import numpy as np
 
def volatility_20d(close_series: pd.Series) -> float:
    """20日年化波动率（基于日收益率标准差）"""
    returns = close_series.pct_change().dropna().tail(20)
    if len(returns) < 5:
        return None
    daily_vol = returns.std()
    annual_vol = daily_vol * np.sqrt(252)  # 年化
    return annual_vol

def calc_beta(stock_returns: pd.Series, benchmark_returns: pd.Series, window=60) -> float:
    """计算个股相对基准（沪深300）的 Beta"""
    aligned = pd.concat([stock_returns, benchmark_returns], axis=1).dropna().tail(window)
    if len(aligned) < 30:
        return None
    covariance = aligned.cov().iloc[0, 1]
    variance = benchmark_returns.var()
    if variance == 0:
        return None
    return covariance / variance

def turnover_anomaly(turnover_series: pd.Series) -> float:
    """当前换手率相对 20 日均值的偏离度"""
    if len(turnover_series) < 21:
        return None
    current = turnover_series.iloc[-1]
    avg_20 = turnover_series.iloc[-21:-1].mean()
    if avg_20 == 0:
        return None
    return current / avg_20 - 1  # >0 表示异常活跃，<0 表示异常冷淡

from scipy import stats
 
def normalize_factors(factor_df: pd.DataFrame) -> pd.DataFrame:
    """
    对全市场同一交易日的因子值做横截面标准化（Z-score）。
    缺失值跳过，不填充。
    """
    result = factor_df.copy()
    factor_cols = [
        'pe', 'pb', 'momentum_20d', 'momentum_60d',
        'roe', 'gross_margin', 'volatility_20d', 'beta',
        'turnover_anomaly', 'netprofit_yoy'
    ]
    
    for col in factor_cols:
        if col in result.columns:
            result[f'{col}_zscore'] = stats.zscore(result[col], nan_policy='omit')
    
    return result
 
def compute_composite_score(row, weights: dict) -> float | None:
    """
    根据标准化因子值和权重计算综合评分。
    缺失因子跳过，权重归一化到剩余因子。
    
    注意：部分因子方向需要反转：
    - PE/PB 越低越好 → zscore 取反
    - 波动率越低越好 → zscore 取反
    """
    direction = {
        'pe_zscore': -1,        # PE 低 = 好
        'pb_zscore': -1,        # PB 低 = 好
        'volatility_20d_zscore': -1,  # 波动低 = 好
        'beta_zscore': -1,      # Beta 低 = 防御性强
        # 其他因子方向为正（越高越好）
    }
    
    available = {}
    available_weights = 0
    
    for factor, weight in weights.items():
        z_col = f'{factor}_zscore'
        if z_col in row.index and not pd.isna(row[z_col]):
            dir_mult = direction.get(z_col, 1)
            available[factor] = row[z_col] * dir_mult * weight
            available_weights += weight
    
    if available_weights == 0:
        return None
    
    return sum(available.values()) / available_weights

# config/factors.yaml
factors:
  weights:
    pe: 0.15
    pb: 0.10
    momentum_20d: 0.15
    momentum_60d: 0.10
    roe: 0.10
    gross_margin: 0.05
    volatility_20d: 0.10
    beta: 0.10
    turnover_anomaly: 0.05
    netprofit_yoy: 0.10
 
  # 因子缺失容忍度：至少需要多少权重覆盖才生成评分
  min_weight_coverage: 0.5  # 50%
  
  # 方向定义：哪些因子值越低越好
  inverse_direction:
    - pe
    - pb
    - volatility_20d
    - beta

def generate_signal(row) -> dict:
    """
    基于单日的所有指标和因子值，生成交易信号。
    返回：{ signal_type, confidence, reasons[] }
    """
    reasons = []
    score = 0
    
    # 1. 综合因子评分贡献
    if row.get('composite_score') is not None:
        z = row['composite_score']
        if z > 1.5:
            score += 3
            reasons.append(f'多因子综合评分显著偏高 (z={z:.2f})')
        elif z > 0.5:
            score += 1
            reasons.append(f'多因子综合评分略高 (z={z:.2f})')
        elif z < -1.5:
            score -= 3
            reasons.append(f'多因子综合评分显著偏低 (z={z:.2f})')
        elif z < -0.5:
            score -= 1
            reasons.append(f'多因子综合评分略低 (z={z:.2f})')
    
    # 2. MACD 信号
    if row.get('macd_hist') is not None:
        if row['macd_hist'] > 0 and row.get('macd_hist_prev', 0) <= 0:
            score += 2
            reasons.append('MACD 金叉（柱转正）')
        elif row['macd_hist'] < 0 and row.get('macd_hist_prev', 0) >= 0:
            score -= 2
            reasons.append('MACD 死叉（柱转负）')
    
    # 3. 均线排列
    if row.get('ma_bullish'):
        score += 1
        reasons.append('均线多头排列')
    
    # 4. RSI 极端值
    if row.get('rsi14') is not None:
        if row['rsi14'] > 80:
            score -= 1
            reasons.append(f'RSI 严重超买 ({row["rsi14"]:.0f})')
        elif row['rsi14'] > 70:
            score -= 0.5
        elif row['rsi14'] < 20:
            score += 1
            reasons.append(f'RSI 严重超卖 ({row["rsi14"]:.0f})')
    
    # 5. 布林带位置
    if row.get('close') and row.get('boll_lower'):
        if row['close'] <= row['boll_lower']:
            score += 0.5
            reasons.append('价格触及布林下轨')
        elif row['close'] >= row['boll_upper']:
            score -= 0.5
    
    # 6. 综合判定
    if score >= 4:
        signal = 'STRONG_BUY'
        confidence = min(0.95, 0.60 + score * 0.05)
    elif score >= 1:
        signal = 'BUY'
        confidence = min(0.80, 0.50 + score * 0.05)
    elif score >= -1:
        signal = 'HOLD'
        confidence = 0.50
    elif score >= -3:
        signal = 'SELL'
        confidence = min(0.80, 0.50 + abs(score) * 0.05)
    else:
        signal = 'STRONG_SELL'
        confidence = min(0.95, 0.60 + abs(score) * 0.05)
    
    return {
        'signal_type': signal,
        'confidence': round(confidence, 2),
        'score': score,
        'reasons': reasons,
    }

CREATE TABLE trade_signal_daily (
    id BIGINT IDENTITY(1,1) PRIMARY KEY,
    stock_code VARCHAR(20) NOT NULL,
    trade_date DATE NOT NULL,
    
    signal_type VARCHAR(20) NOT NULL,  -- STRONG_BUY/BUY/HOLD/SELL/STRONG_SELL
    confidence DECIMAL(4,3),
    composite_score DECIMAL(8,4),      -- 综合因子评分(Z-score)
    signal_score INT,                   -- 算分结果
    reasons NVARCHAR(MAX),              -- JSON: [{"code":"macd_golden","text":"..."}]
    raw_indicators NVARCHAR(MAX),       -- JSON: 触发信号的原始指标值
    
    created_at DATETIME2 DEFAULT GETDATE(),
    
    CONSTRAINT uq_signal_date UNIQUE (stock_code, trade_date)
);
 
CREATE INDEX idx_signal_date ON trade_signal_daily(trade_date);
CREATE INDEX idx_signal_type ON trade_signal_daily(trade_date, signal_type);

from concurrent.futures import ThreadPoolExecutor, as_completed
 
def compute_daily_indicators_and_signals():
    """每日收盘后运行的量化分析任务"""
    
    # 1. 获取全市场股票列表
    stocks = get_all_a_stocks()
    
    # 2. 批量加载行情数据（一次 SQL 查所有股票最近 120 日）
    df_all = fetch_recent_kline_all(back_days=120)
    
    # 3. 按股票分组并行计算
    results = []
    with ThreadPoolExecutor(max_workers=8) as executor:
        futures = {
            executor.submit(compute_one_stock, code, group): code
            for code, group in df_all.groupby('stock_code')
        }
        for future in as_completed(futures):
            try:
                result = future.result()
                results.append(result)
            except Exception as e:
                stock = futures[future]
                logger.error("indicator_compute_failed", stock_code=stock, error=str(e))
    
    # 4. 批量写入
    indicators_df = pd.concat([r['indicators'] for r in results if r])
    signals_df = pd.concat([r['signals'] for r in results if r])
    
    batch_upsert_indicators(indicators_df)
    batch_upsert_signals(signals_df)
    
    # 5. 输出 TOP 信号
    top_buy = signals_df.nlargest(20, 'composite_score')
    top_sell = signals_df.nsmallest(20, 'composite_score')
    
    logger.info("daily_indicators_complete",
        total_stocks=len(stocks),
        indicators_written=len(indicators_df),
        signals_generated=len(signals_df),
        top_buy_signals=len(top_buy),
        top_sell_signals=len(top_sell)
    )
 
def compute_one_stock(code: str, df: pd.DataFrame) -> dict:
    """单只股票：指标 + 因子 + 信号"""
    df = df.sort_values('trade_date')
    
    # 计算技术指标
    df = compute_all_indicators(df)
    
    # 只取今日的数据
    today = df.iloc[-1:]
    if today.empty:
        return None
    
    # 计算今日因子值 + 信号
    signal = generate_signal(today.iloc[0])
    
    return {
        'indicators': today[['stock_code', 'trade_date', 'ma5', 'ma10', 'ma20', 'ma60', 'ma120',
                              'macd_dif', 'macd_dea', 'macd_hist', 'rsi6', 'rsi14', 'rsi24',
                              'kdj_k', 'kdj_d', 'kdj_j', 'boll_upper', 'boll_middle', 'boll_lower',
                              'atr14', 'obv', 'ma_bullish']],
        'signals': pd.DataFrame([{
            'stock_code': code,
            'trade_date': today.iloc[0]['trade_date'],
            **signal,
        }]),
    }

GET /api/v1/indicators/{stock_code}

{
  "stock_code": "600519.SH",
  "data": [
    {
      "trade_date": "2026-07-18",
      "ma5": 1680.5,
      "ma10": 1672.3,
      "rsi14": 62.5,
      "macd_hist": 3.21,
      "ma_bullish": 1
    }
  ]
}

GET /api/v1/signals/daily

{
  "trade_date": "2026-07-18",
  "total": 1523,
  "signals": [
    {
      "stock_code": "600519.SH",
      "stock_name": "贵州茅台",
      "signal_type": "BUY",
      "confidence": 0.72,
      "composite_score": 0.85,
      "reasons": ["MACD 金叉", "多因子综合评分略高"]
    }
  ]
}

GET /api/v1/factors/ranking

# 使用 pandas-ta 的 strategy 模式批量计算
df.ta.strategy(
    ta.CommonStrategy,  # 内置策略：SMA + EMA + MACD + RSI + BBANDS + ATR + OBV
    length=[5, 10, 20, 60, 120]
)

┌─────────────────────────────────────────────────────┐
│  📊 量化分析         [2026-07-18] [沪深300 ▼]        │
├─────────────────────────────────────────────────────┤
│                                                     │
│  ┌─ TOP 买入信号 ──────────────────────────────┐    │
│  │ #1 贵州茅台  BUY  置信度 0.72  Z-score 0.85 │    │
│  │ #2 宁德时代  STRONG_BUY  0.88  1.62         │    │
│  │ #3 ...                                      │    │
│  └──────────────────────────────────────────────┘    │
│                                                     │
│  ┌─ K线图 + 技术指标叠加 ───────────────────────┐   │
│  │ 📈 [K线] [成交量]                             │   │
│  │    叠加: MA5/MA10/MA20                       │   │
│  │    副图: MACD / RSI / KDJ (可切换)            │   │
│  │    标注: 信号点 (买卖标记)                     │   │
│  └──────────────────────────────────────────────┘   │
│                                                     │
│  ┌─ 因子雷达图 ──┐  ┌─ 信号分布 ────────────────┐  │
│  │  (六维雷达)    │  │ 📊 柱状图                   │  │
│  │               │  │ BUY: 320  SELL: 180        │  │
│  └───────────────┘  │ HOLD: 4230                 │  │
│                     └─────────────────────────────┘  │
└─────────────────────────────────────────────────────┘