Trading Quant

{
  "ownerId": "local",
  "slug": "trading-quant",
  "version": "1.0.0",
  "publishedAt": 1772117606781
}

__pycache__/
*.py[cod]
.DS_Store
*.log
venv/
*.bak.*
cache/

pandas>=2.0.0
pandas-ta>=0.3.14b
httpx>=0.27.0
aiohttp>=3.9.0
pyyaml>=6.0
python-dateutil>=2.9.0

#!/bin/bash
# Parallel batch data fetching for cron tasks
# Usage: bash batch_fetch.sh [profile]
# Profiles: closing (收盘), daily (日报), weekly (周报), morning (晨报), macro

PYTHON=/opt/homebrew/bin/python3.12
QUANT=/Users/study/.openclaw/workspace-trading/skills/trading-quant/scripts/quant.py
OUT=/tmp/quant_batch_$$
mkdir -p "$OUT"

profile=${1:-closing}

case "$profile" in
  closing)
    $PYTHON "$QUANT" stock_analysis > "$OUT/stock.json" 2>/dev/null &
    $PYTHON "$QUANT" northbound_flow > "$OUT/northbound.json" 2>/dev/null &
    $PYTHON "$QUANT" market_anomaly > "$OUT/anomaly.json" 2>/dev/null &
    $PYTHON "$QUANT" news_sentiment > "$OUT/news.json" 2>/dev/null &
    $PYTHON "$QUANT" gold_analysis > "$OUT/gold.json" 2>/dev/null &
    $PYTHON "$QUANT" margin_data > "$OUT/margin.json" 2>/dev/null &
    $PYTHON "$QUANT" lhb > "$OUT/lhb.json" 2>/dev/null &
    $PYTHON "$QUANT" top_amount > "$OUT/top_amount.json" 2>/dev/null &
    $PYTHON "$QUANT" save_daily > "$OUT/save.json" 2>/dev/null &
    ;;
  daily)
    $PYTHON "$QUANT" stock_analysis > "$OUT/stock.json" 2>/dev/null &
    $PYTHON "$QUANT" northbound_flow > "$OUT/northbound.json" 2>/dev/null &
    $PYTHON "$QUANT" global_overview > "$OUT/global.json" 2>/dev/null &
    ;;
  weekly)
    $PYTHON "$QUANT" stock_analysis > "$OUT/stock.json" 2>/dev/null &
    $PYTHON "$QUANT" global_overview > "$OUT/global.json" 2>/dev/null &
    ;;
  morning)
    $PYTHON "$QUANT" global_overview > "$OUT/global.json" 2>/dev/null &
    $PYTHON "$QUANT" northbound_flow > "$OUT/northbound.json" 2>/dev/null &
    ;;
  macro)
    $PYTHON "$QUANT" global_overview > "$OUT/global.json" 2>/dev/null &
    $PYTHON "$QUANT" northbound_flow > "$OUT/northbound.json" 2>/dev/null &
    ;;
esac

wait

echo "{"
first=1
for f in "$OUT"/*.json; do
  name=$(basename "$f" .json)
  [ $first -eq 0 ] && echo ","
  echo "\"$name\": $(cat "$f")"
  first=0
done
echo "}"

rm -rf "$OUT"

"""Capital flow analysis — volume/turnover/bid-ask based scoring."""

from __future__ import annotations

import logging
from dataclasses import dataclass, field

from data_sources.base import QuoteData

logger = logging.getLogger(__name__)


@dataclass
class CapitalSignal:
    """Capital flow analysis result."""
    score: float = 50.0
    signals: list[str] = field(default_factory=list)
    metrics: dict = field(default_factory=dict)


def compute_capital(quote: QuoteData, avg_volume: float = 0, avg_amount: float = 0, main_force_data: dict = None) -> CapitalSignal:
    """Compute capital flow score from real-time quote data.

    Uses volume ratio, turnover rate, bid-ask spread, and amount anomaly.
    avg_volume/avg_amount: 5-day average for comparison.
    
    Args:
        main_force_data: 主力资金数据 (来自主力接口)
            {
                "main_net_inflow_wan": -10653,  # 主力净流入 (万)
                "super_big_net_wan": -15104,    # 超大单
                "big_net_wan": 4451,            # 大单
                "signal": "主力流出"
            }
    """
    score = 50.0
    signals = []
    metrics = {}
    chg_pct = quote.change_pct if quote.change_pct is not None else 0.0

    # Volume ratio (量比)
    vr = quote.volume_ratio if quote.volume_ratio is not None else 0
    if vr > 0:
        metrics["volume_ratio"] = round(vr, 2)
        if vr > 5:
            score += 8
            signals.append(f"量比{vr:.1f}极度放量+8")
        elif vr > 3:
            score += 5
            signals.append(f"量比{vr:.1f}显著放量+5")
        elif vr > 1.5:
            score += 2
            signals.append(f"量比{vr:.1f}温和放量+2")
        elif vr < 0.5:
            score -= 3
            signals.append(f"量比{vr:.1f}缩量-3")

    # Turnover rate (换手率)
    tr = quote.turnover_rate if quote.turnover_rate is not None else 0
    if tr > 0:
        metrics["turnover_rate"] = round(tr, 2)
        if tr > 15:
            score += 3
            signals.append(f"换手率{tr:.1f}%高度活跃+3")
        elif tr > 8:
            score += 1
            signals.append(f"换手率{tr:.1f}%活跃+1")
        elif tr < 1:
            score -= 2
            signals.append(f"换手率{tr:.1f}%低迷-2")

    # Amount comparison (成交额对比)
    if avg_amount > 0 and quote.amount > 0:
        amount_ratio = quote.amount / avg_amount
        metrics["amount_ratio"] = round(amount_ratio, 2)
        if amount_ratio > 3:
            score += 5
            signals.append(f"成交额{amount_ratio:.1f}倍均值+5")
        elif amount_ratio > 1.5:
            score += 2
            signals.append(f"成交额{amount_ratio:.1f}倍均值+2")
        elif amount_ratio < 0.5:
            score -= 2
            signals.append(f"成交额仅{amount_ratio:.1f}倍均值-2")

    # Bid-ask pressure (买卖压力)
    if quote.bid1 > 0 and quote.ask1 > 0:
        spread = (quote.ask1 - quote.bid1) / quote.bid1 * 100
        metrics["spread_pct"] = round(spread, 3)
        if spread < 0.05:
            score += 2
            signals.append("买卖价差极窄 (流动性好)+2")

    # Price-volume divergence: price up but volume down => warning
    if chg_pct > 2 and vr > 0 and vr < 0.8:
        score -= 4
        signals.append(f"涨{chg_pct:.1f}%但缩量 (量价背离)-4")
    elif chg_pct < -2 and vr > 3:
        score -= 3
        signals.append(f"跌{chg_pct:.1f}%且放量 (资金出逃)-3")

    # 量比 + 方向修正：放量上涨谨慎，放量下跌警惕
    if vr > 2:
        if chg_pct > 3:
            score -= 3
            signals.append(f"放量大涨 (量比{vr:.1f})-3(追高风险)")
        elif chg_pct < -3:
            score -= 5
            signals.append(f"放量大跌 (量比{vr:.1f})-5(出逃信号)")
        elif chg_pct > 1:
            signals.append(f"放量上涨 (量比{vr:.1f},温和)")

    # 主力资金净流入 (新增)
    if main_force_data and "error" not in main_force_data:
        main_net = main_force_data.get("main_net_inflow_wan", 0)
        super_big = main_force_data.get("super_big_net_wan", 0)
        big = main_force_data.get("big_net_wan", 0)
        signal = main_force_data.get("signal", "")
        source = main_force_data.get("source", "unknown")
        
        metrics["main_net_inflow_wan"] = round(main_net, 2)
        metrics["super_big_net_wan"] = round(super_big, 2)
        metrics["big_net_wan"] = round(big, 2)
        metrics["main_force_source"] = source
        
        # 主力净流入评分
        if main_net > 50000:  # >5 亿
            score += 15
            signals.append(f"主力巨额流入+{main_net/10000:.2f}亿+15[{source}]")
        elif main_net > 5000:  # >5000 万
            score += 10
            signals.append(f"主力大幅流入+{main_net/10000:.2f}亿+10[{source}]")
        elif main_net > 1000:  # >1000 万
            score += 5
            signals.append(f"主力流入+{main_net/10000:.2f}亿+5[{source}]")
        elif main_net < -50000:
            score -= 15
            signals.append(f"主力巨额流出{main_net/10000:.2f}亿-15[{source}]")
        elif main_net < -5000:
            score -= 10
            signals.append(f"主力大幅流出{main_net/10000:.2f}亿-10[{source}]")
        elif main_net < -1000:
            score -= 5
            signals.append(f"主力流出{main_net/10000:.2f}亿-5[{source}]")
        else:
            signals.append(f"主力中性 ({main_net/10000:.2f}亿)[{source}]")
        
        # 超大单修正
        if super_big > 3000:
            score += 3
            signals.append(f"超大单流入+{super_big/10000:.2f}亿+3[{source}]")
        elif super_big < -3000:
            score -= 3
            signals.append(f"超大单流出{super_big/10000:.2f}亿-3[{source}]")
    # 腾讯内外盘差作为主力替代信号
    elif hasattr(quote, 'outer_vol') and hasattr(quote, 'inner_vol'):
        outer_vol = getattr(quote, 'outer_vol', 0) or 0
        inner_vol = getattr(quote, 'inner_vol', 0) or 0
        
        if outer_vol > 0 or inner_vol > 0:
            outer_inner_diff = outer_vol - inner_vol  # 手
            metrics["outer_vol"] = int(outer_vol)
            metrics["inner_vol"] = int(inner_vol)
            metrics["outer_inner_diff"] = int(outer_inner_diff)
            metrics["main_force_source"] = "tencent_outer_inner"
            
            # 内外盘差评分映射 (考虑量比修正)
            base_score = 0
            if outer_inner_diff > 100000:  # >10 万手
                base_score = 8
                signals.append(f"外盘强势 +{outer_inner_diff/10000:.1f}万手+8[腾讯]")
            elif outer_inner_diff > 50000:  # >5 万手
                base_score = 5
                signals.append(f"外盘偏强 +{outer_inner_diff/10000:.1f}万手+5[腾讯]")
            elif outer_inner_diff < -100000:
                base_score = -8
                signals.append(f"内盘强势 {outer_inner_diff/10000:.1f}万手 -8[腾讯]")
            elif outer_inner_diff < -50000:
                base_score = -5
                signals.append(f"内盘偏强 {outer_inner_diff/10000:.1f}万手 -5[腾讯]")
            else:
                signals.append(f"内外盘平衡 (差{outer_inner_diff/10000:.1f}万手)[腾讯]")
            
            # 量比修正 (放量更可信)
            if vr > 3:
                base_score = int(base_score * 1.3)
                signals.append(f"显著放量，可信度 +30%")
            elif vr > 1.5:
                base_score = int(base_score * 1.1)
                signals.append(f"温和放量，可信度 +10%")
            elif vr < 0.8:
                base_score = int(base_score * 0.7)
                signals.append(f"缩量，可信度 -30%")
            
            score += base_score
        else:
            metrics["main_force_source"] = "missing"
            signals.append("主力资金数据缺失 (东方财富不稳定，AKShare 未安装)")
    else:
        # 主力数据缺失，标注数据来源状态
        metrics["main_force_source"] = "missing"
        signals.append("主力资金数据缺失 (东方财富不稳定，AKShare 未安装)")

    return CapitalSignal(
        score=max(0, min(100, score)),
        signals=signals,
        metrics=metrics,
    )

"""Capital flow analysis — volume/turnover/bid-ask based scoring."""

from __future__ import annotations

import logging
from dataclasses import dataclass, field

from data_sources.base import QuoteData

logger = logging.getLogger(__name__)


@dataclass
class CapitalSignal:
    """Capital flow analysis result."""
    score: float = 50.0
    signals: list[str] = field(default_factory=list)
    metrics: dict = field(default_factory=dict)


def compute_capital(quote: QuoteData, avg_volume: float = 0, avg_amount: float = 0, main_force_data: dict = None) -> CapitalSignal:
    """Compute capital flow score from real-time quote data.

    Uses volume ratio, turnover rate, bid-ask spread, and amount anomaly.
    avg_volume/avg_amount: 5-day average for comparison.
    
    Args:
        main_force_data: 主力资金数据 (来自主力接口)
            {
                "main_net_inflow_wan": -10653,  # 主力净流入 (万)
                "super_big_net_wan": -15104,    # 超大单
                "big_net_wan": 4451,            # 大单
                "signal": "主力流出"
            }
    """
    score = 50.0
    signals = []
    metrics = {}
    chg_pct = quote.change_pct if quote.change_pct is not None else 0.0

    # Volume ratio (量比)
    vr = quote.volume_ratio if quote.volume_ratio is not None else 0
    if vr > 0:
        metrics["volume_ratio"] = round(vr, 2)
        if vr > 5:
            score += 8
            signals.append(f"量比{vr:.1f}极度放量+8")
        elif vr > 3:
            score += 5
            signals.append(f"量比{vr:.1f}显著放量+5")
        elif vr > 1.5:
            score += 2
            signals.append(f"量比{vr:.1f}温和放量+2")
        elif vr < 0.5:
            score -= 3
            signals.append(f"量比{vr:.1f}缩量-3")

    # Turnover rate (换手率)
    tr = quote.turnover_rate
    if tr > 0:
        metrics["turnover_rate"] = round(tr, 2)
        if tr > 15:
            score += 3
            signals.append(f"换手率{tr:.1f}%高度活跃+3")
        elif tr > 8:
            score += 1
            signals.append(f"换手率{tr:.1f}%活跃+1")
        elif tr < 1:
            score -= 2
            signals.append(f"换手率{tr:.1f}%低迷-2")

    # Amount comparison (成交额对比)
    if avg_amount > 0 and quote.amount > 0:
        amount_ratio = quote.amount / avg_amount
        metrics["amount_ratio"] = round(amount_ratio, 2)
        if amount_ratio > 3:
            score += 5
            signals.append(f"成交额{amount_ratio:.1f}倍均值+5")
        elif amount_ratio > 1.5:
            score += 2
            signals.append(f"成交额{amount_ratio:.1f}倍均值+2")
        elif amount_ratio < 0.5:
            score -= 2
            signals.append(f"成交额仅{amount_ratio:.1f}倍均值-2")

    # Bid-ask pressure (买卖压力)
    if quote.bid1 > 0 and quote.ask1 > 0:
        spread = (quote.ask1 - quote.bid1) / quote.bid1 * 100
        metrics["spread_pct"] = round(spread, 3)
        if spread < 0.05:
            score += 2
            signals.append("买卖价差极窄 (流动性好)+2")

    # Price-volume divergence: price up but volume down => warning
    if chg_pct > 2 and vr > 0 and vr < 0.8:
        score -= 4
        signals.append(f"涨{chg_pct:.1f}%但缩量 (量价背离)-4")
    elif chg_pct < -2 and vr > 3:
        score -= 3
        signals.append(f"跌{chg_pct:.1f}%且放量 (资金出逃)-3")

    # 量比 + 方向修正：放量上涨谨慎，放量下跌警惕
    if quote.volume_ratio > 2:
        if chg_pct > 3:
            score -= 3
            signals.append(f"放量大涨 (量比{quote.volume_ratio:.1f})-3(追高风险)")
        elif chg_pct < -3:
            score -= 5
            signals.append(f"放量大跌 (量比{quote.volume_ratio:.1f})-5(出逃信号)")
        elif chg_pct > 1:
            signals.append(f"放量上涨 (量比{quote.volume_ratio:.1f},温和)")

    # 主力资金净流入 (新增)
    if main_force_data and "error" not in main_force_data:
        main_net = main_force_data.get("main_net_inflow_wan", 0)
        super_big = main_force_data.get("super_big_net_wan", 0)
        big = main_force_data.get("big_net_wan", 0)
        signal = main_force_data.get("signal", "")
        source = main_force_data.get("source", "unknown")
        
        metrics["main_net_inflow_wan"] = round(main_net, 2)
        metrics["super_big_net_wan"] = round(super_big, 2)
        metrics["big_net_wan"] = round(big, 2)
        metrics["main_force_source"] = source
        
        # 主力净流入评分
        if main_net > 50000:  # >5 亿
            score += 15
            signals.append(f"主力巨额流入+{main_net/10000:.2f}亿+15[{source}]")
        elif main_net > 5000:  # >5000 万
            score += 10
            signals.append(f"主力大幅流入+{main_net/10000:.2f}亿+10[{source}]")
        elif main_net > 1000:  # >1000 万
            score += 5
            signals.append(f"主力流入+{main_net/10000:.2f}亿+5[{source}]")
        elif main_net < -50000:
            score -= 15
            signals.append(f"主力巨额流出{main_net/10000:.2f}亿-15[{source}]")
        elif main_net < -5000:
            score -= 10
            signals.append(f"主力大幅流出{main_net/10000:.2f}亿-10[{source}]")
        elif main_net < -1000:
            score -= 5
            signals.append(f"主力流出{main_net/10000:.2f}亿-5[{source}]")
        else:
            signals.append(f"主力中性 ({main_net/10000:.2f}亿)[{source}]")
        
        # 超大单修正
        if super_big > 3000:
            score += 3
            signals.append(f"超大单流入+{super_big/10000:.2f}亿+3[{source}]")
        elif super_big < -3000:
            score -= 3
            signals.append(f"超大单流出{super_big/10000:.2f}亿-3[{source}]")
    # 腾讯内外盘差作为主力替代信号
    elif hasattr(quote, 'outer_vol') and hasattr(quote, 'inner_vol'):
        outer_vol = getattr(quote, 'outer_vol', 0) or 0
        inner_vol = getattr(quote, 'inner_vol', 0) or 0
        
        if outer_vol > 0 or inner_vol > 0:
            outer_inner_diff = outer_vol - inner_vol  # 手
            metrics["outer_vol"] = int(outer_vol)
            metrics["inner_vol"] = int(inner_vol)
            metrics["outer_inner_diff"] = int(outer_inner_diff)
            metrics["main_force_source"] = "tencent_outer_inner"
            
            # 内外盘差评分映射 (考虑量比修正)
            base_score = 0
            if outer_inner_diff > 100000:  # >10 万手
                base_score = 8
                signals.append(f"外盘强势 +{outer_inner_diff/10000:.1f}万手+8[腾讯]")
            elif outer_inner_diff > 50000:  # >5 万手
                base_score = 5
                signals.append(f"外盘偏强 +{outer_inner_diff/10000:.1f}万手+5[腾讯]")
            elif outer_inner_diff < -100000:
                base_score = -8
                signals.append(f"内盘强势 {outer_inner_diff/10000:.1f}万手 -8[腾讯]")
            elif outer_inner_diff < -50000:
                base_score = -5
                signals.append(f"内盘偏强 {outer_inner_diff/10000:.1f}万手 -5[腾讯]")
            else:
                signals.append(f"内外盘平衡 (差{outer_inner_diff/10000:.1f}万手)[腾讯]")
            
            # 量比修正 (放量更可信)
            if quote.volume_ratio > 3:
                base_score = int(base_score * 1.3)
                signals.append(f"显著放量，可信度 +30%")
            elif quote.volume_ratio > 1.5:
                base_score = int(base_score * 1.1)
                signals.append(f"温和放量，可信度 +10%")
            elif quote.volume_ratio < 0.8:
                base_score = int(base_score * 0.7)
                signals.append(f"缩量，可信度 -30%")
            
            score += base_score
        else:
            metrics["main_force_source"] = "missing"
            signals.append("主力资金数据缺失 (东方财富不稳定，AKShare 未安装)")
    else:
        # 主力数据缺失，标注数据来源状态
        metrics["main_force_source"] = "missing"
        signals.append("主力资金数据缺失 (东方财富不稳定，AKShare 未安装)")

    return CapitalSignal(
        score=max(0, min(100, score)),
        signals=signals,
        metrics=metrics,
    )

"""Capital flow analysis — volume/turnover/bid-ask based scoring."""

from __future__ import annotations

import logging
from dataclasses import dataclass, field

from data_sources.base import QuoteData

logger = logging.getLogger(__name__)


@dataclass
class CapitalSignal:
    """Capital flow analysis result."""
    score: float = 50.0
    signals: list[str] = field(default_factory=list)
    metrics: dict = field(default_factory=dict)


def compute_capital(quote: QuoteData, avg_volume: float = 0, avg_amount: float = 0, main_force_data: dict = None) -> CapitalSignal:
    """Compute capital flow score from real-time quote data.

    Uses volume ratio, turnover rate, bid-ask spread, and amount anomaly.
    avg_volume/avg_amount: 5-day average for comparison.
    
    Args:
        main_force_data: 主力资金数据 (来自主力接口)
            {
                "main_net_inflow_wan": -10653,  # 主力净流入 (万)
                "super_big_net_wan": -15104,    # 超大单
                "big_net_wan": 4451,            # 大单
                "signal": "主力流出"
            }
    """
    score = 50.0
    signals = []
    metrics = {}
    chg_pct = quote.change_pct if quote.change_pct is not None else 0.0

    # Volume ratio (量比)
    vr = quote.volume_ratio if quote.volume_ratio is not None else 0
    if vr > 0:
        metrics["volume_ratio"] = round(vr, 2)
        if vr > 5:
            score += 8
            signals.append(f"量比{vr:.1f}极度放量+8")
        elif vr > 3:
            score += 5
            signals.append(f"量比{vr:.1f}显著放量+5")
        elif vr > 1.5:
            score += 2
            signals.append(f"量比{vr:.1f}温和放量+2")
        elif vr < 0.5:
            score -= 3
            signals.append(f"量比{vr:.1f}缩量-3")

    # Turnover rate (换手率)
    tr = quote.turnover_rate if quote.turnover_rate is not None else 0
    if tr > 0:
        metrics["turnover_rate"] = round(tr, 2)
        if tr > 15:
            score += 3
            signals.append(f"换手率{tr:.1f}%高度活跃+3")
        elif tr > 8:
            score += 1
            signals.append(f"换手率{tr:.1f}%活跃+1")
        elif tr < 1:
            score -= 2
            signals.append(f"换手率{tr:.1f}%低迷-2")

    # Amount comparison (成交额对比)
    if avg_amount > 0 and quote.amount > 0:
        amount_ratio = quote.amount / avg_amount
        metrics["amount_ratio"] = round(amount_ratio, 2)
        if amount_ratio > 3:
            score += 5
            signals.append(f"成交额{amount_ratio:.1f}倍均值+5")
        elif amount_ratio > 1.5:
            score += 2
            signals.append(f"成交额{amount_ratio:.1f}倍均值+2")
        elif amount_ratio < 0.5:
            score -= 2
            signals.append(f"成交额仅{amount_ratio:.1f}倍均值-2")

    # Bid-ask pressure (买卖压力)
    if quote.bid1 > 0 and quote.ask1 > 0:
        spread = (quote.ask1 - quote.bid1) / quote.bid1 * 100
        metrics["spread_pct"] = round(spread, 3)
        if spread < 0.05:
            score += 2
            signals.append("买卖价差极窄 (流动性好)+2")

    # Price-volume divergence: price up but volume down => warning
    if chg_pct > 2 and vr > 0 and vr < 0.8:
        score -= 4
        signals.append(f"涨{chg_pct:.1f}%但缩量 (量价背离)-4")
    elif chg_pct < -2 and vr > 3:
        score -= 3
        signals.append(f"跌{chg_pct:.1f}%且放量 (资金出逃)-3")

    # 量比 + 方向修正：放量上涨谨慎，放量下跌警惕
    if quote.volume_ratio > 2:
        if chg_pct > 3:
            score -= 3
            signals.append(f"放量大涨 (量比{quote.volume_ratio:.1f})-3(追高风险)")
        elif chg_pct < -3:
            score -= 5
            signals.append(f"放量大跌 (量比{quote.volume_ratio:.1f})-5(出逃信号)")
        elif chg_pct > 1:
            signals.append(f"放量上涨 (量比{quote.volume_ratio:.1f},温和)")

    # 主力资金净流入 (新增)
    if main_force_data and "error" not in main_force_data:
        main_net = main_force_data.get("main_net_inflow_wan", 0)
        super_big = main_force_data.get("super_big_net_wan", 0)
        big = main_force_data.get("big_net_wan", 0)
        signal = main_force_data.get("signal", "")
        source = main_force_data.get("source", "unknown")
        
        metrics["main_net_inflow_wan"] = round(main_net, 2)
        metrics["super_big_net_wan"] = round(super_big, 2)
        metrics["big_net_wan"] = round(big, 2)
        metrics["main_force_source"] = source
        
        # 主力净流入评分
        if main_net > 50000:  # >5 亿
            score += 15
            signals.append(f"主力巨额流入+{main_net/10000:.2f}亿+15[{source}]")
        elif main_net > 5000:  # >5000 万
            score += 10
            signals.append(f"主力大幅流入+{main_net/10000:.2f}亿+10[{source}]")
        elif main_net > 1000:  # >1000 万
            score += 5
            signals.append(f"主力流入+{main_net/10000:.2f}亿+5[{source}]")
        elif main_net < -50000:
            score -= 15
            signals.append(f"主力巨额流出{main_net/10000:.2f}亿-15[{source}]")
        elif main_net < -5000:
            score -= 10
            signals.append(f"主力大幅流出{main_net/10000:.2f}亿-10[{source}]")
        elif main_net < -1000:
            score -= 5
            signals.append(f"主力流出{main_net/10000:.2f}亿-5[{source}]")
        else:
            signals.append(f"主力中性 ({main_net/10000:.2f}亿)[{source}]")
        
        # 超大单修正
        if super_big > 3000:
            score += 3
            signals.append(f"超大单流入+{super_big/10000:.2f}亿+3[{source}]")
        elif super_big < -3000:
            score -= 3
            signals.append(f"超大单流出{super_big/10000:.2f}亿-3[{source}]")
    # 腾讯内外盘差作为主力替代信号
    elif hasattr(quote, 'outer_vol') and hasattr(quote, 'inner_vol'):
        outer_vol = getattr(quote, 'outer_vol', 0) or 0
        inner_vol = getattr(quote, 'inner_vol', 0) or 0
        
        if outer_vol > 0 or inner_vol > 0:
            outer_inner_diff = outer_vol - inner_vol  # 手
            metrics["outer_vol"] = int(outer_vol)
            metrics["inner_vol"] = int(inner_vol)
            metrics["outer_inner_diff"] = int(outer_inner_diff)
            metrics["main_force_source"] = "tencent_outer_inner"
            
            # 内外盘差评分映射 (考虑量比修正)
            base_score = 0
            if outer_inner_diff > 100000:  # >10 万手
                base_score = 8
                signals.append(f"外盘强势 +{outer_inner_diff/10000:.1f}万手+8[腾讯]")
            elif outer_inner_diff > 50000:  # >5 万手
                base_score = 5
                signals.append(f"外盘偏强 +{outer_inner_diff/10000:.1f}万手+5[腾讯]")
            elif outer_inner_diff < -100000:
                base_score = -8
                signals.append(f"内盘强势 {outer_inner_diff/10000:.1f}万手 -8[腾讯]")
            elif outer_inner_diff < -50000:
                base_score = -5
                signals.append(f"内盘偏强 {outer_inner_diff/10000:.1f}万手 -5[腾讯]")
            else:
                signals.append(f"内外盘平衡 (差{outer_inner_diff/10000:.1f}万手)[腾讯]")
            
            # 量比修正 (放量更可信)
            if quote.volume_ratio > 3:
                base_score = int(base_score * 1.3)
                signals.append(f"显著放量，可信度 +30%")
            elif quote.volume_ratio > 1.5:
                base_score = int(base_score * 1.1)
                signals.append(f"温和放量，可信度 +10%")
            elif quote.volume_ratio < 0.8:
                base_score = int(base_score * 0.7)
                signals.append(f"缩量，可信度 -30%")
            
            score += base_score
        else:
            metrics["main_force_source"] = "missing"
            signals.append("主力资金数据缺失 (东方财富不稳定，AKShare 未安装)")
    else:
        # 主力数据缺失，标注数据来源状态
        metrics["main_force_source"] = "missing"
        signals.append("主力资金数据缺失 (东方财富不稳定，AKShare 未安装)")

    return CapitalSignal(
        score=max(0, min(100, score)),
        signals=signals,
        metrics=metrics,
    )

"""行业分类模块 - 完全依赖券商接口获取准确数据."""

import json
import logging
from typing import Optional, Dict

logger = logging.getLogger(__name__)

# 持久化缓存行业数据
_CACHE_FILE = "/tmp/quant_industry_cache.json"
_industry_cache: Dict[str, str] = {}
_cache_loaded = False

def _load_file_cache():
    global _industry_cache, _cache_loaded
    if _cache_loaded:
        return
    try:
        import os
        if os.path.exists(_CACHE_FILE):
            with open(_CACHE_FILE) as f:
                _industry_cache = json.load(f)
    except Exception:
        pass
    if not _industry_cache:
        _prefill_from_watchlist()
    _cache_loaded = True


def _prefill_from_watchlist():
    """从 watchlist.json 预填充行业分类（自选股行业不变，无需网络请求）"""
    global _industry_cache
    try:
        import os
        wl_path = os.path.join(os.path.dirname(__file__), "..", "..", "..", "..", "..", "knowledge", "watchlist.json")
        wl_path = os.path.normpath(wl_path)
        if not os.path.exists(wl_path):
            return
        with open(wl_path) as f:
            wl = json.load(f)
        for cat in ("priority", "observe", "research"):
            for item in wl.get(cat, []):
                if isinstance(item, dict):
                    code = item.get("code", "")
                    sector = item.get("sector", "")
                    if code and sector:
                        _industry_cache[code] = _normalize_industry_name(sector.split("/")[0])
        if _industry_cache:
            _save_file_cache()
            logger.info(f"Pre-filled {len(_industry_cache)} industries from watchlist")
    except Exception as e:
        logger.debug(f"Watchlist prefill failed: {e}")

def _save_file_cache():
    try:
        import json as _json
        with open(_CACHE_FILE, "w") as f:
            _json.dump(_industry_cache, f, ensure_ascii=False)
    except Exception:
        pass


def classify_industry(code: str) -> str:
    """
    使用券商/数据接口获取准确的行业分类.
    
    优先级:
    1. 东方财富接口 (eastmoney) - 最可靠
    2. akshare 股票基本信息
    3. 同花顺接口 (ths)
    
    如果所有接口都失败，返回 "unknown"，不进行猜测.
    
    Args:
        code: 股票代码
    
    Returns:
        行业分类名称，失败返回 "unknown"
    """
    _load_file_cache()
    if code in _industry_cache:
        return _industry_cache[code]
    
    industry = None
    
    # 尝试 1: 东方财富接口 (akshare)
    try:
        industry = _get_industry_from_eastmoney(code)
        if industry and industry != "unknown":
            _industry_cache[code] = industry
            _save_file_cache()
            logger.debug(f"Got industry for {code} from EastMoney: {industry}")
            return industry
    except Exception as e:
        logger.debug(f"EastMoney industry fetch failed for {code}: {e}")
    
    # 尝试 2: akshare 股票基本信息（拉取全市场数据，性能差，降级跳过）
    # _get_industry_from_akshare 会调用 stock_zh_a_spot_em() 拉取 5000+ 股票
    # 东方财富个股接口已足够，无需全市场查询
    
    # 尝试 3: 同花顺接口（非常慢，遍历所有行业板块，降级跳过）
    # 注释掉以避免性能问题：对每只股票遍历所有行业板块是O(N*M)复杂度
    # try:
    #     industry = _get_industry_from_ths(code)
    #     ...
    
    # 所有接口都失败，返回 unknown
    logger.warning(f"Failed to get industry for {code} from all sources")
    _industry_cache[code] = "unknown"
    _save_file_cache()
    return "unknown"


def _get_industry_from_eastmoney(code: str) -> Optional[str]:
    """从东方财富获取行业分类（通过akshare）."""
    try:
        import akshare as ak
        
        # 使用东方财富个股信息接口
        df = ak.stock_individual_info_em(symbol=code)
        if df is not None and not df.empty:
            # 查找行业信息
            industry_row = df[df['item'] == '行业']
            if not industry_row.empty:
                industry = industry_row['value'].values[0]
                if industry and industry != '-' and industry != 'None':
                    return _normalize_industry_name(industry)
        
        return "unknown"
    except Exception as e:
        logger.debug(f"Failed to get EastMoney industry for {code}: {e}")
        return "unknown"


def _get_industry_from_akshare(code: str) -> Optional[str]:
    """从 akshare 获取行业分类."""
    try:
        import akshare as ak
        
        # 使用股票基本信息接口
        df = ak.stock_zh_a_spot_em()
        if df is not None and not df.empty:
            stock_row = df[df['代码'] == code]
            if not stock_row.empty:
                # 尝试获取行业列
                if '行业' in stock_row.columns:
                    industry = stock_row['行业'].values[0]
                    if industry and industry != '-' and pd.notna(industry):
                        return _normalize_industry_name(industry)
        
        return "unknown"
    except Exception as e:
        logger.debug(f"Failed to get AKShare industry for {code}: {e}")
        return "unknown"


def _get_industry_from_ths(code: str) -> Optional[str]:
    """从同花顺获取行业分类."""
    try:
        import akshare as ak
        
        # 使用同花顺行业数据接口
        # 获取所有行业板块
        df = ak.stock_board_industry_name_ths()
        if df is not None and not df.empty:
            # 遍历行业，查找成分股
            for _, row in df.iterrows():
                industry_name = row.get('名称', '')
                if not industry_name:
                    continue
                
                try:
                    # 获取该行业的成分股
                    stocks_df = ak.stock_board_industry_cons_ths(symbol=industry_name)
                    if stocks_df is not None and not stocks_df.empty:
                        # 检查目标股票是否在该行业
                        if '代码' in stocks_df.columns:
                            if code in stocks_df['代码'].values:
                                return _normalize_industry_name(industry_name)
                        elif 'symbol' in stocks_df.columns:
                            if code in stocks_df['symbol'].values:
                                return _normalize_industry_name(industry_name)
                except Exception:
                    continue
        
        return "unknown"
    except Exception as e:
        logger.debug(f"Failed to get THS industry for {code}: {e}")
        return "unknown"


def _normalize_industry_name(industry: str) -> str:
    """
    标准化行业名称到内部分类.
    
    将各种数据源返回的行业名称统一为内部标准名称.
    """
    if not industry or industry == '-':
        return "unknown"
    
    industry = str(industry).strip()
    
    # 直接映射表
    industry_map = {
        # 银行
        "银行": "银行", "商业银行": "银行", "股份制银行": "银行", "城商行": "银行",
        
        # 地产
        "房地产": "地产", "房地产开发": "地产", "物业管理": "地产", "商业地产": "地产",
        
        # 钢铁
        "钢铁": "钢铁", "普钢": "钢铁", "特钢": "钢铁", "钢铁制品": "钢铁",
        
        # 煤炭
        "煤炭": "煤炭", "煤炭开采": "煤炭", "焦炭": "煤炭", "煤化工": "煤炭",
        
        # 石油
        "石油": "石油", "石油化工": "石油", "油气开采": "石油", "炼油": "石油",
        
        # 电力
        "电力": "电力", "火电": "电力", "水电": "电力", "核电": "电力", 
        "风电": "新能源", "光伏": "新能源", "电力设备": "新能源",
        
        # 有色
        "有色金属": "有色", "贵金属": "有色", "工业金属": "有色", "稀有金属": "有色",
        "铜": "有色", "铝": "有色", "锌": "有色", "镍": "有色", "钴": "有色", "锂": "有色",
        
        # 化工
        "化工": "化工", "化学制品": "化工", "化学原料": "化工", "化肥": "化工", 
        "农药": "化工", "化纤": "化工", "塑料": "化工", "橡胶": "化工",
        
        # 消费
        "食品加工": "消费", "饮料制造": "消费", "休闲食品": "消费", "调味品": "消费",
        "乳业": "消费", "肉制品": "消费", "啤酒": "消费", "食品饮料": "消费",
        
        # 白酒
        "白酒": "白酒", "白酒Ⅱ": "白酒", "白酒Ⅲ": "白酒", "酿酒": "白酒",
        
        # 医药
        "医药": "医药", "生物医药": "医药", "化学制药": "医药", "中药": "医药", 
        "医疗器械": "医药", "医疗服务": "医药", "疫苗": "医药", "生物制品": "医药",
        
        # 科技
        "计算机": "科技", "软件": "科技", "IT服务": "科技", "互联网": "科技",
        "通信": "科技", "通信设备": "科技", "传媒": "科技", "电子": "半导体",
        
        # 半导体
        "半导体": "半导体", "集成电路": "半导体", "芯片": "半导体", 
        "元件": "半导体", "光学光电子": "半导体",
        
        # 军工
        "国防军工": "军工", "军工": "军工", "航天": "军工", "航空": "军工", 
        "船舶": "军工", "兵器": "军工",
        
        # 新能源
        "新能源": "新能源", "电池": "新能源", "锂电池": "新能源", "储能": "新能源",
        "光伏设备": "新能源", "风电设备": "新能源", "新能源汽车": "新能源",
        "电网设备": "新能源",
        
        # ETF
        "ETF": "ETF",
    }
    
    # 直接映射
    if industry in industry_map:
        return industry_map[industry]
    
    # 部分匹配
    for key, value in industry_map.items():
        if key in industry:
            return value
    
    # 无法识别的行业，返回原始值（小写处理）
    logger.debug(f"Unrecognized industry name: {industry}, using as-is")
    return industry


def clear_industry_cache():
    """清除行业分类缓存."""
    global _industry_cache
    _industry_cache.clear()


def get_cached_industries() -> Dict[str, str]:
    """获取已缓存的行业分类."""
    return _industry_cache.copy()


# 兼容性别名
classify_industry_by_api = classify_industry

"""TradingScore V2 — multi-dimensional stock scoring engine."""

from __future__ import annotations

import logging
from dataclasses import dataclass, field
from typing import Any

from config import get_config
from data_sources.base import QuoteData
from .technical import TechnicalSignal, compute_technical
from .capital_flow import CapitalSignal, compute_capital
from .industry_classifier import classify_industry_by_api, classify_industry

logger = logging.getLogger(__name__)


@dataclass
class StockScore:
    """Complete multi-dimensional score for a stock."""
    code: str
    name: str
    price: float
    change_pct: float
    total_score: float = 50.0
    technical: dict = field(default_factory=dict)
    capital: dict = field(default_factory=dict)
    sentiment: dict = field(default_factory=dict)
    fundamental: dict = field(default_factory=dict)
    market: dict = field(default_factory=dict)
    signal: str = "WATCH"
    confidence: str = "中"
    risk_alerts: list[str] = field(default_factory=list)
    data_freshness: str = "fresh"

    def to_dict(self) -> dict:
        return {
            "code": self.code,
            "name": self.name,
            "price": self.price,
            "change_pct": self.change_pct,
            "score": {
                "total": round(self.total_score, 1),
                "technical": self.technical,
                "capital": self.capital,
                "fundamental": self.fundamental,
                "sentiment": self.sentiment,
                "market": self.market,
            },
            "signal": self.signal,
            "confidence": self.confidence,
            "risk_alerts": self.risk_alerts,
            "data_freshness": self.data_freshness,
        }


def _get_signal(score: float) -> tuple[str, str]:
    """Map score to signal level and confidence."""
    cfg = get_config().get("scoring", {}).get("signals", {})
    if score >= cfg.get("strong_buy", 80):
        return "STRONG_BUY", "高"
    elif score >= cfg.get("buy", 65):
        return "BUY", "中高"
    elif score >= cfg.get("watch", 50):
        return "WATCH", "中"
    elif score >= cfg.get("hold", 35):
        return "HOLD", "中低"
    elif score >= cfg.get("sell", 22):
        return "SELL", "低"
    elif score >= cfg.get("strong_sell", 18):
        return "STRONG_SELL", "极低"
    else:
        return "STRONG_SELL", "极低"






INDUSTRY_PE_RANGES = {
    "银行": {"undervalued": 5, "fair": 8, "high": 12, "extreme": 20},
    "地产": {"undervalued": 5, "fair": 10, "high": 15, "extreme": 25},
    "钢铁": {"undervalued": 5, "fair": 10, "high": 15, "extreme": 25},
    "煤炭": {"undervalued": 5, "fair": 8, "high": 12, "extreme": 20},
    "石油": {"undervalued": 6, "fair": 12, "high": 18, "extreme": 30},
    "电力": {"undervalued": 8, "fair": 15, "high": 25, "extreme": 40},
    "有色": {"undervalued": 8, "fair": 15, "high": 25, "extreme": 45},
    "化工": {"undervalued": 8, "fair": 15, "high": 25, "extreme": 40},
    "消费": {"undervalued": 15, "fair": 25, "high": 40, "extreme": 60},
    "白酒": {"undervalued": 15, "fair": 30, "high": 45, "extreme": 70},
    "医药": {"undervalued": 15, "fair": 30, "high": 50, "extreme": 80},
    "科技": {"undervalued": 15, "fair": 35, "high": 60, "extreme": 100},
    "半导体": {"undervalued": 15, "fair": 40, "high": 70, "extreme": 120},
    "军工": {"undervalued": 20, "fair": 40, "high": 70, "extreme": 100},
    "新能源": {"undervalued": 15, "fair": 30, "high": 50, "extreme": 80},
    "ETF": {"undervalued": 0, "fair": 999, "high": 999, "extreme": 999},
    "default": {"undervalued": 10, "fair": 20, "high": 40, "extreme": 80},
}

INDUSTRY_KEYWORDS = {
    "银行": ["银行", "Bank"],
    "地产": ["地产", "置业", "置地"],
    "钢铁": ["钢铁", "钢材", "不锈钢"],
    "煤炭": ["煤炭", "煤业", "能源"],
    "石油": ["石油", "石化", "中海油"],
    "电力": ["电力", "电网", "发电"],
    "有色": ["有色", "铜", "铝", "锌", "镍", "锡", "铅", "银锡", "矿业", "紫金"],
    "化工": ["化工", "化学"],
    "消费": ["食品", "饮料", "乳业", "调味"],
    "白酒": ["茅台", "五粮液", "洋河", "泸州", "汾酒", "酒"],
    "医药": ["医药", "生物", "制药", "疫苗"],
    "科技": ["科技", "软件", "信息", "数据", "AI", "智能"],
    "半导体": ["芯片", "半导体", "存储", "光刻"],
    "军工": ["军工", "航天", "航空", "兵器", "船舶", "导弹", "国防"],
    "新能源": ["光伏", "风电", "锂电", "新能源", "电池", "储能", "金风"],
    "ETF": ["ETF"],
}


# 保留旧函数别名以兼容现有代码，但标记为废弃
def _classify_industry(name: str) -> str:
    """
    [废弃] 根据股票名称推断行业分类.
    
    请使用 classify_industry_by_api(code, name) 替代.
    """
    from .industry_classifier import _classify_industry_by_name
    return _classify_industry_by_name(name)


def _get_pe_ranges(industry: str) -> dict:
    """获取行业PE合理区间."""
    return INDUSTRY_PE_RANGES.get(industry, INDUSTRY_PE_RANGES["default"])


def _compute_fundamental(quote: QuoteData) -> dict:
    """Compute fundamental score from PE/PB data."""
    score = 50.0
    signals = []

    pe = quote.pe if quote.pe is not None else 0
    pb = quote.pb if quote.pb is not None else 0

    if pe <= 0:
        signals.append("PE数据缺失(中性)")
        return {"score": score, "signals": signals, "pe": pe, "pb": pb}

    # PE scoring with industry classification
    industry = classify_industry(quote.code)
    pe_ranges = _get_pe_ranges(industry)
    
    if pe < pe_ranges["undervalued"]:
        score += 15
        signals.append(f"PE={pe:.1f}(低估值,{industry})")
    elif pe < pe_ranges["fair"]:
        score += 8
        signals.append(f"PE={pe:.1f}(合理,{industry})")
    elif pe < pe_ranges["high"]:
        signals.append(f"PE={pe:.1f}(偏高,{industry})")
    elif pe < pe_ranges["extreme"]:
        score -= 8
        signals.append(f"PE={pe:.1f}(高估,{industry})")
    else:
        score -= 15
        signals.append(f"PE={pe:.1f}(极高估,{industry})")

    # PB scoring
    if pb > 0:
        if pb < 1:
            score += 10
            signals.append(f"PB={pb:.2f}(破净)")
        elif pb < 2:
            score += 5
            signals.append(f"PB={pb:.2f}(低PB)")
        elif pb > 10:
            score -= 5
            signals.append(f"PB={pb:.2f}(高PB)")

    score = max(0, min(100, score))
    return {"score": round(score, 1), "signals": signals, "pe": pe, "pb": pb, "market_cap": quote.market_cap}

def compute_stock_score(
    quote: QuoteData,
    daily_df: Any,
    avg_volume: float = 0,
    avg_amount: float = 0,
    extra: dict = None,
    capital_flow_data: dict = None,  # 新增：主力资金数据
) -> StockScore:
    """Compute TradingScore V2 for a single stock.

    Combines technical (30%), capital (35%), sentiment (20%), market (15%).
    Sentiment and market scores default to neutral when data unavailable.
    
    Args:
        capital_flow_data: 主力资金数据 (来自主力接口或龙虎榜)
    """
    if extra is None:
        extra = {}
    change_pct = quote.change_pct if quote.change_pct is not None else 0.0
    cfg = get_config().get("scoring", {}).get("weights", {})
    w_tech = cfg.get("technical", 0.25)
    w_cap = cfg.get("capital", 0.30)
    w_fund = cfg.get("fundamental", 0.10)
    w_sent = cfg.get("sentiment", 0.20)
    w_mkt = cfg.get("market", 0.15)

    w_total = w_tech + w_cap + w_fund + w_sent + w_mkt
    if abs(w_total - 1.0) > 0.001:
        w_tech /= w_total
        w_cap /= w_total
        w_fund /= w_total
        w_sent /= w_total
        w_mkt /= w_total

    tech = compute_technical(daily_df)
    cap = compute_capital(quote, avg_volume=avg_volume, avg_amount=avg_amount, main_force_data=capital_flow_data)
    fund = _compute_fundamental(quote)

    sent_score = 50.0
    sent_signals = []
    news_sentiment = extra.get("news_sentiment", None)
    if news_sentiment is not None:
        news_count = extra.get("news_count", 0)
        # Map news sentiment (-5 to +5) to score (30 to 70)
        sent_score = 50.0 + news_sentiment * 4.0
        sent_score = max(30.0, min(70.0, sent_score))
        if news_sentiment > 0.5:
            sent_signals.append(f"新闻偏多({news_count}条,情绪+{news_sentiment})")
        elif news_sentiment < -0.5:
            sent_signals.append(f"新闻偏空({news_count}条,情绪{news_sentiment})")
        else:
            sent_signals.append(f"新闻中性({news_count}条)")
        top_news = extra.get("top_news", [])
        for t in top_news[:1]:
            sent_signals.append(f"热点:{t[:20]}")
    else:
        sent_signals.append("消息面暂无数据(中性)")

    mkt_data = extra.get("market_sentiment", None)
    if mkt_data and isinstance(mkt_data, dict):
        mkt_score = float(mkt_data.get("score", 50.0))
        mkt_signals = mkt_data.get("signals", ["市场情绪数据加载中"])
        
        # 大盘连涨天数惩罚(从extra中获取)
        consecutive_up_days = extra.get("consecutive_up_days", 0)
        if consecutive_up_days >= 5:
            mkt_score -= 10
            mkt_signals.append(f"大盘连涨{consecutive_up_days}天(过热风险)-10")
        elif consecutive_up_days >= 3:
            mkt_score -= 5
            mkt_signals.append(f"大盘连涨{consecutive_up_days}天(谨慎)-5")
        
        consecutive_down_days = extra.get("consecutive_down_days", 0)
        if consecutive_down_days >= 5:
            mkt_score += 8
            mkt_signals.append(f"大盘连跌{consecutive_down_days}天(超跌反弹概率)+8")
        elif consecutive_down_days >= 3:
            mkt_score += 4
            mkt_signals.append(f"大盘连跌{consecutive_down_days}天(关注底部)+4")
        
        # 北向资金连续净流出
        nb_consecutive_outflow = extra.get("nb_consecutive_outflow_days", 0)
        if nb_consecutive_outflow >= 5:
            mkt_score -= 8
            mkt_signals.append(f"北向连续净流出{nb_consecutive_outflow}天(外资撤退)-8")
        elif nb_consecutive_outflow >= 3:
            mkt_score -= 4
            mkt_signals.append(f"北向连续净流出{nb_consecutive_outflow}天(外资谨慎)-4")
        
        mkt_score = max(15, min(85, mkt_score))
    else:
        mkt_score = 50.0
        mkt_signals = ["市场情绪暂无数据(中性)"]

    total = (
        tech.score * w_tech
        + cap.score * w_cap
        + fund["score"] * w_fund
        + sent_score * w_sent
        + mkt_score * w_mkt
    )

    # 防追涨杀跌: 动量惩罚 (涨幅越大, 越可能是追涨)
    momentum_penalty = 0
    chg = abs(change_pct)
    if change_pct >= 9.5:
        momentum_penalty = 12
    elif change_pct >= 7:
        momentum_penalty = 6
    elif change_pct >= 5:
        momentum_penalty = 3
    elif change_pct <= -9.5:
        momentum_penalty = -8  # 跌停时逆向加分(超跌反弹概率)
    elif change_pct <= -7:
        momentum_penalty = -4
    
    if momentum_penalty != 0:
        total -= momentum_penalty  # 涨多扣分, 跌多加分(逆向)
    total = max(10, min(90, total))

    risk_alerts = _detect_risks(quote, tech, cap, change_pct)
    signal, confidence = _get_signal(total)
    
    # RSI超买时限制最高信号为WATCH
    rsi_val = tech.indicators.get("rsi", 50)
    if rsi_val > 80 and signal in ("STRONG_BUY", "BUY"):
        signal = "WATCH"
        confidence = "中"
        risk_alerts.append(f"RSI={rsi_val:.0f}超买,信号降级为WATCH")
    
    # RSI超卖时保底信号为WATCH
    if rsi_val < 20 and signal in ("STRONG_SELL", "SELL"):
        signal = "WATCH"
        confidence = "中"
        risk_alerts.append(f"RSI={rsi_val:.0f}超卖,可能超跌反弹")

    data_freshness = "fresh"
    if quote.timestamp:
        data_freshness = "fresh"
    elif daily_df is not None and not daily_df.empty:
        data_freshness = "stale"
    else:
        data_freshness = "cached"

    return StockScore(
        code=quote.code,
        name=quote.name,
        price=quote.price,
        change_pct=change_pct,
        total_score=round(total, 1),
        technical={"score": round(tech.score, 1), "signals": tech.signals, "indicators": tech.indicators},
        capital={"score": round(cap.score, 1), "signals": cap.signals, "metrics": cap.metrics},
        fundamental={"score": fund["score"], "signals": fund["signals"], "pe": fund.get("pe", 0), "pb": fund.get("pb", 0)},
        sentiment={"score": sent_score, "signals": sent_signals},
        market={"score": mkt_score, "signals": mkt_signals},
        signal=signal,
        confidence=confidence,
        risk_alerts=risk_alerts,
        data_freshness=data_freshness,
    )


def _detect_risks(quote: QuoteData, tech: TechnicalSignal, cap: CapitalSignal, change_pct: float = 0.0) -> list[str]:
    """Detect risk conditions from combined analysis."""
    alerts = []

    if change_pct >= 9.5:
        alerts.append(f"涨停{change_pct:.1f}%(追高风险)")
    elif change_pct >= 7:
        alerts.append(f"大涨{change_pct:.1f}%(短期回调风险)")
    elif change_pct <= -9.5:
        alerts.append(f"跌停{change_pct:.1f}%(恐慌信号)")
    elif change_pct <= -5:
        alerts.append(f"大跌{change_pct:.1f}%(止损关注)")

    rsi = tech.indicators.get("rsi", 50)
    if rsi > 80:
        alerts.append(f"RSI={rsi:.0f}(极度超买)")
    elif rsi < 20:
        alerts.append(f"RSI={rsi:.0f}(极度超卖)")

    vr = cap.metrics.get("volume_ratio", 1)
    if vr > 5 and change_pct > 5:
        alerts.append(f"量比{vr:.1f}+涨幅{change_pct:.1f}%(游资炒作可能)")

    if hasattr(quote, "pe") and (quote.pe or 0) > 100:
        alerts.append(f"PE={quote.pe:.0f}(极高估值风险)")
    if hasattr(quote, "pb") and quote.pb is not None and 0 < quote.pb < 0.8:
        alerts.append(f"PB={quote.pb:.2f}(破净，注意基本面)")

    return alerts

"""金融情感分析模块 - 基于 FinBERT.

该模块提供基于 FinBERT 的金融文本情感分析功能。
支持单条和批量分析，包含延迟加载和 LLM fallback 机制。
"""

import os
import json
from typing import List, Dict, Union, Optional
from dataclasses import dataclass, asdict
from enum import Enum
import logging

logger = logging.getLogger(__name__)


class SentimentLabel(str, Enum):
    """情感标签枚举"""
    POSITIVE = "positive"
    NEGATIVE = "negative"
    NEUTRAL = "neutral"
    ERROR = "error"
    EMPTY = "empty"


@dataclass
class SentimentResult:
    """情感分析结果.

    Attributes:
        score: 情感分数，范围 -1.0（极度负面）到 1.0（极度正面）
        label: 情感标签（positive/negative/neutral/error/empty）
        confidence: 模型置信度，范围 0.0 到 1.0
        raw_label: 原始模型输出标签
        method: 分析方法（finbert/llm_fallback/rule_based/empty）
    """
    score: float = 0.0
    label: str = "neutral"
    confidence: float = 0.0
    raw_label: str = ""
    method: str = "unknown"

    def to_dict(self) -> Dict:
        """转换为字典."""
        return asdict(self)

    def is_bullish(self, threshold: float = 0.1) -> bool:
        """判断是否看涨.

        Args:
            threshold: 看涨阈值，默认 0.1

        Returns:
            如果情感分数大于阈值返回 True
        """
        return self.score > threshold

    def is_bearish(self, threshold: float = -0.1) -> bool:
        """判断是否看跌.

        Args:
            threshold: 看跌阈值，默认 -0.1

        Returns:
            如果情感分数小于阈值返回 True
        """
        return self.score < threshold


class FinBERTSentimentAnalyzer:
    """FinBERT 情感分析器.

    使用中文金融新闻 FinBERT 模型进行情感分析。
    支持延迟加载、批量处理和多种 fallback 策略。

    Example:
        >>> analyzer = FinBERTSentimentAnalyzer()
        >>> result = analyzer.analyze("央行降息，股市上涨")
        >>> print(result.score, result.label)
    """

    # 中文金融新闻 FinBERT 模型
    DEFAULT_MODEL = "uer/roberta-base-finetuned-chinanews-chinese"

    # 备用模型（按优先级排序）
    FALLBACK_MODELS = [
        "uer/roberta-base-finetuned-jd-binary-chinese",
        "uer/roberta-base-finetuned-dianping-chinese",
        "uer/roberta-base-finetuned-weibo-chinese",
    ]

    # 情感关键词映射（rule-based fallback）
    BULLISH_KEYWORDS = [
        "上涨", "涨停", "利好", "增长", "盈利", "突破", "买入", "增持",
        "反弹", "创新高", "超预期", "降息", "降准", "刺激", "支持",
        "利好", "景气", "扩张", "回升", "改善", "复苏", "牛市",
        "rise", "surge", "rally", "bull", "gain", "profit", "growth",
        "upgrade", "buy", "outperform", "beat", "positive"
    ]

    BEARISH_KEYWORDS = [
        "下跌", "跌停", "利空", "亏损", "跌破", "卖出", "减持",
        "回调", "创新低", "不及预期", "加息", "收紧", "风险", "危机",
        "衰退", "萎缩", "下滑", "恶化", "熊市", "panic", "crash",
        "fall", "drop", "decline", "bear", "loss", "sell", "downgrade",
        "underperform", "miss", "negative", "warning"
    ]

    def __init__(
        self,
        model_name: Optional[str] = None,
        device: str = "cpu",
        use_mps: bool = False,
        cache_dir: Optional[str] = None
    ):
        """初始化情感分析器.

        Args:
            model_name: 模型名称，默认使用中文 FinBERT
            device: 运行设备，"cpu"、"cuda" 或 "mps"
            use_mps: 是否尝试使用 Apple Silicon MPS
            cache_dir: HuggingFace 缓存目录
        """
        self.model_name = model_name or self.DEFAULT_MODEL
        self.device = self._select_device(device, use_mps)
        self.cache_dir = cache_dir or os.path.expanduser("~/.cache/huggingface")

        self._pipeline = None
        self._model_loaded = False
        self._model_info = {
            "model_name": self.model_name,
            "loaded": False,
            "device": self.device,
            "cache_dir": self.cache_dir,
            "fallback_used": False
        }

    def _select_device(self, device: str, use_mps: bool) -> str:
        """选择最佳可用设备.

        Args:
            device: 用户指定的设备
            use_mps: 是否允许使用 MPS

        Returns:
            实际使用的设备名称
        """
        if device != "auto":
            return device

        try:
            import torch
            if torch.cuda.is_available():
                return "cuda"
            if use_mps and torch.backends.mps.is_available():
                return "mps"
        except ImportError:
            pass

        return "cpu"

    def _load_model(self) -> bool:
        """延迟加载模型.

        Returns:
            是否成功加载模型
        """
        if self._model_loaded:
            return True

        try:
            from transformers import pipeline, AutoTokenizer, AutoModelForSequenceClassification

            logger.info(f"Loading FinBERT model: {self.model_name}")

            # 尝试加载主模型
            loaded = self._try_load_model(
                self.model_name,
                AutoTokenizer,
                AutoModelForSequenceClassification,
                pipeline
            )

            # 主模型失败，尝试备用模型
            if not loaded:
                for fallback_model in self.FALLBACK_MODELS:
                    logger.warning(f"Trying fallback model: {fallback_model}")
                    loaded = self._try_load_model(
                        fallback_model,
                        AutoTokenizer,
                        AutoModelForSequenceClassification,
                        pipeline
                    )
                    if loaded:
                        self._model_info["model_name"] = fallback_model
                        self._model_info["fallback_used"] = True
                        break

            if loaded:
                self._model_loaded = True
                self._model_info["loaded"] = True
                logger.info(f"Model loaded successfully on {self.device}")
                return True
            else:
                logger.error("All models failed to load")
                return False

        except ImportError as e:
            logger.warning(f"transformers not installed: {e}")
            return False
        except Exception as e:
            logger.error(f"Failed to load model: {e}")
            return False

    def _try_load_model(
        self,
        model_name: str,
        AutoTokenizer,
        AutoModelForSequenceClassification,
        pipeline
    ) -> bool:
        """尝试加载单个模型.

        Args:
            model_name: 模型名称
            AutoTokenizer: Tokenizer 类
            AutoModelForSequenceClassification: 模型类

        Returns:
            是否成功加载
        """
        try:
            # 尝试从本地缓存加载
            try:
                tokenizer = AutoTokenizer.from_pretrained(
                    model_name,
                    local_files_only=True,
                    cache_dir=self.cache_dir
                )
                model = AutoModelForSequenceClassification.from_pretrained(
                    model_name,
                    local_files_only=True,
                    cache_dir=self.cache_dir
                )
                logger.info(f"Model loaded from local cache: {model_name}")
            except (OSError, ValueError):
                # 从 HuggingFace 下载
                logger.info(f"Downloading model: {model_name}")
                tokenizer = AutoTokenizer.from_pretrained(
                    model_name,
                    cache_dir=self.cache_dir
                )
                model = AutoModelForSequenceClassification.from_pretrained(
                    model_name,
                    cache_dir=self.cache_dir
                )

            # 确定设备映射
            device_id = -1  # CPU default
            if self.device == "mps":
                import torch
                device_id = 0 if torch.backends.mps.is_available() else -1
            elif self.device == "cuda":
                import torch
                device_id = 0 if torch.cuda.is_available() else -1

            self._pipeline = pipeline(
                "sentiment-analysis",
                model=model,
                tokenizer=tokenizer,
                device=device_id,
                truncation=True,
                max_length=512
            )

            return True

        except Exception as e:
            logger.warning(f"Failed to load {model_name}: {e}")
            return False

    def analyze(self, text: str) -> SentimentResult:
        """分析单条文本情感.

        Args:
            text: 待分析文本（新闻标题、摘要等）

        Returns:
            SentimentResult 包含标准化后的情感分数
        """
        if not text or not text.strip():
            return SentimentResult(
                score=0.0,
                label=SentimentLabel.EMPTY,
                confidence=0.0,
                raw_label="",
                method="empty"
            )

        if not self._load_model():
            # 模型加载失败，使用 rule-based fallback
            return self._rule_based_fallback(text)

        try:
            result = self._pipeline(text[:512])
            raw = result[0]

            label = raw['label'].lower()
            confidence = raw['score']

            # 映射到 -1.0 ~ 1.0
            if 'positive' in label or label == 'positive':
                score = confidence
                normalized_label = SentimentLabel.POSITIVE
            elif 'negative' in label or label == 'negative':
                score = -confidence
                normalized_label = SentimentLabel.NEGATIVE
            else:
                score = 0.0
                normalized_label = SentimentLabel.NEUTRAL

            return SentimentResult(
                score=round(score, 3),
                label=normalized_label,
                confidence=round(confidence, 3),
                raw_label=label,
                method="finbert"
            )

        except Exception as e:
            logger.error(f"FinBERT analysis failed: {e}")
            return self._rule_based_fallback(text)

    def analyze_batch(self, texts: List[str]) -> List[SentimentResult]:
        """批量分析文本情感（效率更高）.

        Args:
            texts: 待分析文本列表

        Returns:
            SentimentResult 列表，与输入顺序一致
        """
        if not texts:
            return []

        if not self._load_model():
            return [self._rule_based_fallback(t) for t in texts]

        try:
            # 记录原始索引，过滤空文本
            valid_indices = []
            valid_texts = []
            for i, t in enumerate(texts):
                if t and t.strip():
                    valid_indices.append(i)
                    valid_texts.append(t[:512])

            if not valid_texts:
                return [
                    SentimentResult(0.0, SentimentLabel.EMPTY, 0.0, "", "empty")
                    for _ in texts
                ]

            results = self._pipeline(valid_texts)

            # 构建完整结果列表
            processed = [
                SentimentResult(0.0, SentimentLabel.EMPTY, 0.0, "", "empty")
                for _ in texts
            ]

            for idx, raw in zip(valid_indices, results):
                label = raw['label'].lower()
                confidence = raw['score']

                if 'positive' in label:
                    score = confidence
                    norm_label = SentimentLabel.POSITIVE
                elif 'negative' in label:
                    score = -confidence
                    norm_label = SentimentLabel.NEGATIVE
                else:
                    score = 0.0
                    norm_label = SentimentLabel.NEUTRAL

                processed[idx] = SentimentResult(
                    score=round(score, 3),
                    label=norm_label,
                    confidence=round(confidence, 3),
                    raw_label=label,
                    method="finbert"
                )

            return processed

        except Exception as e:
            logger.error(f"Batch analysis failed: {e}")
            return [self._rule_based_fallback(t) for t in texts]

    def _rule_based_fallback(self, text: str) -> SentimentResult:
        """基于规则的情感分析 fallback.

        当 FinBERT 不可用时，使用关键词匹配进行简单情感判断。

        Args:
            text: 待分析文本

        Returns:
            SentimentResult
        """
        if not text:
            return SentimentResult(0.0, SentimentLabel.EMPTY, 0.0, "", "empty")

        text_lower = text.lower()

        bullish_count = sum(1 for kw in self.BULLISH_KEYWORDS if kw in text_lower)
        bearish_count = sum(1 for kw in self.BEARISH_KEYWORDS if kw in text_lower)

        total = bullish_count + bearish_count
        if total == 0:
            return SentimentResult(0.0, SentimentLabel.NEUTRAL, 0.5, "neutral", "rule_based")

        # 计算情感分数
        score = (bullish_count - bearish_count) / max(total, 3)
        score = max(-1.0, min(1.0, score))  # 限制在 -1 到 1

        if score > 0.1:
            label = SentimentLabel.POSITIVE
        elif score < -0.1:
            label = SentimentLabel.NEGATIVE
        else:
            label = SentimentLabel.NEUTRAL

        confidence = min(0.7, total / 10)  # 关键词越多置信度越高，最高 0.7

        return SentimentResult(
            score=round(score, 3),
            label=label,
            confidence=round(confidence, 3),
            raw_label="keyword_match",
            method="rule_based"
        )

    def get_model_info(self) -> Dict:
        """获取模型信息.

        Returns:
            包含模型状态信息的字典
        """
        return self._model_info.copy()

    def warmup(self) -> bool:
        """预热模型（预加载并执行一次推理）.

        Returns:
            是否成功预热
        """
        success = self._load_model()
        if success:
            # 执行一次虚拟推理确保模型完全加载
            _ = self.analyze("测试")
        return success


# 单例模式 - 全局复用
_sentiment_analyzer: Optional[FinBERTSentimentAnalyzer] = None


def get_sentiment_analyzer(
    model_name: Optional[str] = None,
    device: str = "auto"
) -> FinBERTSentimentAnalyzer:
    """获取全局情感分析器实例（单例模式）.

    Args:
        model_name: 模型名称，None 使用默认
        device: 运行设备

    Returns:
        FinBERTSentimentAnalyzer 实例
    """
    global _sentiment_analyzer
    if _sentiment_analyzer is None:
        _sentiment_analyzer = FinBERTSentimentAnalyzer(
            model_name=model_name,
            device=device
        )
    return _sentiment_analyzer


def reset_sentiment_analyzer():
    """重置全局分析器（用于测试）."""
    global _sentiment_analyzer
    _sentiment_analyzer = None


async def analyze_news_sentiment(
    title: str,
    summary: str = "",
    title_weight: float = 0.7
) -> SentimentResult:
    """分析新闻情感（供数据源调用）.

    优先使用标题，如果有摘要则加权合并分析。

    Args:
        title: 新闻标题
        summary: 新闻摘要
        title_weight: 标题权重（0-1），默认 0.7

    Returns:
        SentimentResult
    """
    analyzer = get_sentiment_analyzer()

    # 只有标题
    if not summary or len(summary) < 10:
        return analyzer.analyze(title)

    # 标题 + 摘要
    title_result = analyzer.analyze(title)
    summary_result = analyzer.analyze(summary[:300])

    # 加权合并
    combined_score = (
        title_result.score * title_weight +
        summary_result.score * (1 - title_weight)
    )

    # 确定最终标签
    if combined_score > 0.1:
        label = SentimentLabel.POSITIVE
    elif combined_score < -0.1:
        label = SentimentLabel.NEGATIVE
    else:
        label = SentimentLabel.NEUTRAL

    # 置信度取平均
    confidence = (title_result.confidence + summary_result.confidence) / 2

    return SentimentResult(
        score=round(combined_score, 3),
        label=label,
        confidence=round(confidence, 3),
        raw_label=f"title:{title_result.raw_label},summary:{summary_result.raw_label}",
        method="finbert_weighted"
    )


def analyze_news_batch(
    news_items: List[Dict],
    text_key: str = "title",
    summary_key: str = "summary"
) -> List[Dict]:
    """批量分析新闻列表.

    Args:
        news_items: 新闻列表，每项是包含 title 和可选 summary 的字典
        text_key: 标题字段名
        summary_key: 摘要字段名

    Returns:
        添加 sentiment 相关字段的 news_items
    """
    analyzer = get_sentiment_analyzer()

    # 构建待分析文本
    texts = []
    for item in news_items:
        title = item.get(text_key, "")
        summary = item.get(summary_key, "")
        if summary and len(summary) > 10:
            text = f"{title}。{summary[:200]}"
        else:
            text = title
        texts.append(text)

    results = analyzer.analyze_batch(texts)

    # 添加结果到原数据
    for item, result in zip(news_items, results):
        item["sentiment"] = result.score
        item["sentiment_label"] = result.label
        item["sentiment_confidence"] = result.confidence
        item["sentiment_method"] = result.method

    return news_items


def calculate_aggregate_sentiment(
    results: List[SentimentResult],
    method: str = "mean"
) -> Dict:
    """计算聚合情感.

    Args:
        results: 情感结果列表
        method: 聚合方法，"mean" 或 "weighted"

    Returns:
        聚合结果字典
    """
    if not results:
        return {
            "score": 0.0,
            "label": "neutral",
            "confidence": 0.0,
            "count": 0
        }

    if method == "weighted":
        # 按置信度加权
        total_weight = sum(r.confidence for r in results)
        if total_weight == 0:
            score = sum(r.score for r in results) / len(results)
        else:
            score = sum(r.score * r.confidence for r in results) / total_weight
    else:
        score = sum(r.score for r in results) / len(results)

    # 确定聚合标签
    if score > 0.1:
        label = "bullish"
    elif score < -0.1:
        label = "bearish"
    else:
        label = "neutral"

    return {
        "score": round(score, 3),
        "label": label,
        "confidence": round(sum(r.confidence for r in results) / len(results), 3),
        "count": len(results),
        "positive_count": sum(1 for r in results if r.label == SentimentLabel.POSITIVE),
        "negative_count": sum(1 for r in results if r.label == SentimentLabel.NEGATIVE),
        "neutral_count": sum(1 for r in results if r.label == SentimentLabel.NEUTRAL),
    }