#!/usr/bin/env python3
"""
analytics_engine.py — 实时 + 批处理分析引擎

职责:
  1. 实时分析: 从新事件流计算即时指标 (窗口滚动)
  2. 批处理分析: 每日/每周聚合统计 (胜率、盈亏、夏普比)
  3. 模式识别: 识别交易模式 (剥头皮、日内、波段)
  4. 趋势检测: 市场状态变化检测

分析维度:
  - 盈利能力: 总盈亏、胜率、平均盈亏比
  - 效率指标: 持仓时长分布、执行延迟分布
  - 风险指标: 最大回撤、连亏次数、风险评分
  - 模式分析: 交易时间段分布、品种分布、regime分布
"""
from __future__ import annotations

import json
import math
import os
import time
from collections import defaultdict, deque
from dataclasses import asdict, dataclass, field
from datetime import datetime, timedelta, timezone
from pathlib import Path
from typing import Any, Optional

# ── 配置 ─────────────────────────────────────────────────────────

APP_DIR = Path(__file__).resolve().parent.parent.parent
CLEANED_DATA_DIR = APP_DIR / "data" / "consumed" / "cleaned"
ANALYTICS_DIR = APP_DIR / "data" / "consumed" / "analytics"
REPORTS_DIR = APP_DIR / "reports"
STATS_DIR = APP_DIR / "logs"

# 滚动窗口配置
REALTIME_WINDOW_MIN = 30  # 30分钟实时窗口
BATCH_INTERVAL_SEC = 300  # 5分钟批处理间隔


# ── 数据类 ───────────────────────────────────────────────────────

@dataclass
class RollingStats:
    """滚动统计 (实时)"""
    window_min: int = 30
    event_count: int = 0
    wins: int = 0
    losses: int = 0
    total_pnl: float = 0.0
    max_pnl: float = 0.0
    min_pnl: float = 0.0
    avg_confidence: float = 0.0
    avg_latency_ms: float = 0.0
    regime_counts: dict[str, int] = field(default_factory=dict)
    side_counts: dict[str, int] = field(default_factory=dict)
    updated_at: str = ""


@dataclass
class BatchReport:
    """批处理分析报告"""
    period_start: str = ""
    period_end: str = ""
    total_events: int = 0
    wins: int = 0
    losses: int = 0
    win_rate: float = 0.0
    total_pnl: float = 0.0
    avg_pnl: float = 0.0
    avg_win: float = 0.0
    avg_loss: float = 0.0
    profit_factor: float = 0.0
    max_drawdown: float = 0.0
    sharpe_ratio: float = 0.0
    avg_confidence: float = 0.0
    avg_latency_ms: float = 0.0
    avg_duration_min: float = 0.0
    regime_distribution: dict[str, int] = field(default_factory=dict)
    side_distribution: dict[str, int] = field(default_factory=dict)
    hourly_distribution: dict[int, int] = field(default_factory=dict)
    execution_quality: dict[str, Any] = field(default_factory=dict)
    generated_at: str = ""


@dataclass
class MarketInsight:
    """市场洞察"""
    insight_type: str = ""  # regime_change / pattern_detected / anomaly
    severity: str = ""       # INFO / WARN / ALERT
    title: str = ""
    description: str = ""
    evidence: dict = field(default_factory=dict)
    recommended_action: str = ""
    ts: str = ""


# ── 实时分析器 ───────────────────────────────────────────────────

class RealTimeAnalyzer:
    """实时滚动分析器 (30分钟窗口)"""

    def __init__(self, window_min: int = 30):
        self.window_min = window_min
        self._events: deque[tuple[datetime, dict]] = deque()
        self._stats = RollingStats(window_min=window_min)

    def _cleanup_window(self, now: datetime) -> None:
        """清理窗口外的过期事件"""
        cutoff = now - timedelta(minutes=self.window_min)
        while self._events and self._events[0][0] < cutoff:
            self._events.popleft()

    def _recompute_stats(self) -> RollingStats:
        """从窗口事件重新计算统计"""
        stats = RollingStats(window_min=self.window_min)
        if not self._events:
            return stats

        pnl_values = []
        confidences = []
        latencies = []

        for _, event in self._events:
            stats.event_count += 1
            pnl = event.get("pnl", 0)
            pnl_values.append(pnl)
            stats.total_pnl += pnl

            if pnl > 0:
                stats.wins += 1
            elif pnl < 0:
                stats.losses += 1

            stats.max_pnl = max(stats.max_pnl, pnl)
            stats.min_pnl = min(stats.min_pnl, pnl)

            regime = event.get("regime", "UNKNOWN")
            stats.regime_counts[regime] = stats.regime_counts.get(regime, 0) + 1

            side = event.get("side", "UNKNOWN")
            stats.side_counts[side] = stats.side_counts.get(side, 0) + 1

            conf = event.get("confidence", 0.5)
            confidences.append(conf)

            latency = event.get("execution_latency_ms", 500)
            latencies.append(latency)

        if pnl_values:
            stats.avg_confidence = sum(confidences) / len(confidences)
            stats.avg_latency_ms = sum(latencies) / len(latencies)
        else:
            stats.avg_confidence = 0.5
            stats.avg_latency_ms = 500

        stats.updated_at = datetime.now(timezone.utc).isoformat()
        self._stats = stats
        return stats

    def add_event(self, event: dict) -> RollingStats:
        """添加新事件并更新统计"""
        ts_str = event.get("ts", "")
        dt = datetime.now(timezone.utc)
        if ts_str:
            for fmt in ("%Y-%m-%dT%H:%M:%S", "%Y-%m-%dT%H:%M:%SZ"):
                try:
                    dt = datetime.strptime(ts_str[:19], fmt).replace(tzinfo=timezone.utc)
                    break
                except ValueError:
                    pass

        self._events.append((dt, event))
        self._cleanup_window(dt)
        return self._recompute_stats()

    def get_stats(self) -> RollingStats:
        """获取当前统计"""
        if self._events:
            self._cleanup_window(datetime.now(timezone.utc))
        return self._recompute_stats()


# ── 批处理分析器 ─────────────────────────────────────────────────

class BatchAnalyzer:
    """周期性批处理分析器"""

    def __init__(self):
        self.realtime = RealTimeAnalyzer(window_min=REALTIME_WINDOW_MIN)

    def analyze_period(
        self,
        events: list[dict],
        period_start: datetime | None = None,
        period_end: datetime | None = None,
    ) -> BatchReport:
        """分析给定时间段的事件"""
        if not events:
            return BatchReport()

        if period_end is None:
            period_end = datetime.now(timezone.utc)
        if period_start is None:
            period_start = period_end - timedelta(hours=24)

        # 过滤时间范围内的事件
        filtered = []
        for e in events:
            ts_str = e.get("ts", "")
            if not ts_str:
                continue
            for fmt in ("%Y-%m-%dT%H:%M:%S", "%Y-%m-%dT%H:%M:%SZ"):
                try:
                    dt = datetime.strptime(ts_str[:19], fmt).replace(tzinfo=timezone.utc)
                    if period_start <= dt <= period_end:
                        filtered.append(e)
                    break
                except ValueError:
                    continue

        return self._compute_report(filtered, period_start, period_end)

    def _compute_report(
        self,
        events: list[dict],
        period_start: datetime,
        period_end: datetime,
    ) -> BatchReport:
        """从事件列表计算报告"""
        report = BatchReport()
        report.period_start = period_start.isoformat()
        report.period_end = period_end.isoformat()
        report.total_events = len(events)

        if not events:
            report.generated_at = datetime.now(timezone.utc).isoformat()
            return report

        win_events = []
        loss_events = []
        pnl_values = []
        confidences = []
        latencies = []
        durations = []
        hourly_counts = defaultdict(int)
        regime_counts = defaultdict(int)
        side_counts = defaultdict(int)

        for e in events:
            pnl = e.get("pnl", 0)
            conf = e.get("confidence", 0.5)
            latency = e.get("execution_latency_ms", 500)
            duration = e.get("duration_min")
            regime = e.get("regime", "UNKNOWN")
            side = e.get("side", "UNKNOWN")

            pnl_values.append(pnl)
            confidences.append(conf)
            latencies.append(latency)
            if duration is not None:
                durations.append(duration)

            regime_counts[regime] += 1
            side_counts[side] += 1

            if pnl > 0:
                win_events.append(e)
                report.wins += 1
            elif pnl < 0:
                loss_events.append(e)
                report.losses += 1

            # 小时分布
            ts_str = e.get("ts", "")
            if ts_str:
                try:
                    dt = datetime.strptime(ts_str[:19], "%Y-%m-%dT%H:%M:%S").replace(tzinfo=timezone.utc)
                    hourly_counts[dt.hour] += 1
                except ValueError:
                    pass

        report.total_pnl = sum(pnl_values)
        report.win_rate = report.wins / report.total_events if report.total_events > 0 else 0
        report.avg_pnl = report.total_pnl / report.total_events if report.total_events > 0 else 0

        if win_events:
            report.avg_win = sum(e.get("pnl", 0) for e in win_events) / len(win_events)
        if loss_events:
            report.avg_loss = sum(abs(e.get("pnl", 0)) for e in loss_events) / len(loss_events)

        # 盈亏比
        if report.avg_loss > 0:
            report.profit_factor = report.avg_win / report.avg_loss

        # 夏普比 (简化版, 无风险利率=0)
        if len(pnl_values) > 1:
            mean_pnl = sum(pnl_values) / len(pnl_values)
            variance = sum((p - mean_pnl) ** 2 for p in pnl_values) / len(pnl_values)
            std_dev = math.sqrt(variance) if variance > 0 else 1
            report.sharpe_ratio = round(mean_pnl / std_dev * math.sqrt(252), 3)
        else:
            report.sharpe_ratio = 0.0

        # 最大回撤
        cumulative = 0
        peak = 0
        max_dd = 0
        for pnl in pnl_values:
            cumulative += pnl
            peak = max(peak, cumulative)
            dd = peak - cumulative
            max_dd = max(max_dd, dd)
        report.max_drawdown = round(max_dd, 2)

        # 平均值
        report.avg_confidence = round(sum(confidences) / len(confidences), 3) if confidences else 0.5
        report.avg_latency_ms = round(sum(latencies) / len(latencies), 1) if latencies else 500
        report.avg_duration_min = round(sum(durations) / len(durations), 1) if durations else 0

        report.regime_distribution = dict(regime_counts)
        report.side_distribution = dict(side_counts)
        report.hourly_distribution = {h: hourly_counts[h] for h in sorted(hourly_counts)}

        # 执行质量评分
        latency_good = sum(1 for l in latencies if l < 200)
        conf_good = sum(1 for c in confidences if c >= 0.7)
        report.execution_quality = {
            "fast_execution_pct": round(latency_good / len(latencies) * 100, 1) if latencies else 0,
            "high_confidence_pct": round(conf_good / len(confidences) * 100, 1) if confidences else 0,
            "overall_score": round(
                (latency_good / len(latencies) * 0.4 + conf_good / len(confidences) * 0.6) * 100, 1
            ) if latencies and confidences else 0,
        }

        report.generated_at = datetime.now(timezone.utc).isoformat()
        return report

    def add_realtime_event(self, event: dict) -> RollingStats:
        """添加实时事件 (供监控面板使用)"""
        return self.realtime.add_event(event)


# ── 模式识别 ─────────────────────────────────────────────────────

class PatternRecognizer:
    """交易模式识别器"""

    PATTERNS = {
        "scalping": {"max_duration_min": 5, "min_count": 3},
        "day_trading": {"min_duration_min": 5, "max_duration_min": 480},
        "swing": {"min_duration_min": 480},
    }

    def detect_pattern(self, events: list[dict]) -> dict[str, Any]:
        """检测交易模式分布"""
        patterns = defaultdict(lambda: {"count": 0, "events": []})

        for e in events:
            duration = e.get("duration_min", 0)
            for name, criteria in self.PATTERNS.items():
                if "max_duration_min" in criteria and duration <= criteria["max_duration_min"]:
                    if "min_duration_min" not in criteria or duration >= criteria["min_duration_min"]:
                        patterns[name]["count"] += 1
                        patterns[name]["events"].append(e)
                        break
            else:
                patterns["other"]["count"] += 1
                patterns[name]["events"].append(e)

        return {k: {"count": v["count"], "pct": round(v["count"] / len(events) * 100, 1) if events else 0}
                for k, v in patterns.items()}

    def detect_regime_change(
        self,
        current_events: list[dict],
        historical_events: list[dict],
    ) -> Optional[MarketInsight]:
        """检测市场状态变化"""
        if len(current_events) < 5 or len(historical_events) < 10:
            return None

        current_regimes = [e.get("regime", "UNKNOWN") for e in current_events]
        historical_regimes = [e.get("regime", "UNKNOWN") for e in historical_events[-50:]]

        # 计算各 regime 占比
        def regime_ratio(regimes, target):
            return regimes.count(target) / len(regimes) if regimes else 0

        current_trend = regime_ratio(current_regimes, "TRENDING")
        hist_trend = regime_ratio(historical_regimes, "TRENDING")
        current_range = regime_ratio(current_regimes, "RANGING")
        hist_range = regime_ratio(historical_regimes, "RANGING")

        # 检测趋势强度变化
        if abs(current_trend - hist_trend) > 0.3:
            direction = "增强" if current_trend > hist_trend else "减弱"
            return MarketInsight(
                insight_type="regime_change",
                severity="WARN" if abs(current_trend - hist_trend) > 0.5 else "INFO",
                title=f"趋势强度{direction}",
                description=f"TRENDING 占比从 {hist_trend:.0%} 变为 {current_trend:.0%}",
                evidence={
                    "current_trend_ratio": round(current_trend, 3),
                    "historical_trend_ratio": round(hist_trend, 3),
                    "current_range_ratio": round(current_range, 3),
                },
                recommended_action="调整策略参数以适应新市场状态",
                ts=datetime.now(timezone.utc).isoformat(),
            )

        return None

    def detect_anomaly(self, events: list[dict], rolling_stats: RollingStats) -> list[MarketInsight]:
        """检测异常模式"""
        insights = []

        # 检测执行延迟异常
        if rolling_stats.avg_latency_ms > 1000:
            insights.append(MarketInsight(
                insight_type="latency_spike",
                severity="ALERT",
                title="执行延迟异常",
                description=f"平均延迟 {rolling_stats.avg_latency_ms:.0f}ms, 超过阈值",
                evidence={"avg_latency_ms": rolling_stats.avg_latency_ms},
                recommended_action="检查网络连接和 broker 状态",
                ts=datetime.now(timezone.utc).isoformat(),
            ))

        # 检测胜率异常
        if rolling_stats.event_count >= 10:
            wr = rolling_stats.wins / rolling_stats.event_count
            if wr < 0.30:
                insights.append(MarketInsight(
                    insight_type="low_winrate",
                    severity="WARN",
                    title="胜率过低",
                    description=f"30分钟内胜率 {wr:.1%}, 建议检查策略参数",
                    evidence={"win_rate": round(wr, 3), "event_count": rolling_stats.event_count},
                    recommended_action="收紧信号置信度或暂停交易",
                    ts=datetime.now(timezone.utc).isoformat(),
                ))

        return insights


# ── 分析引擎主类 ─────────────────────────────────────────────────

class AnalyticsEngine:
    """
    统一分析引擎, 整合实时和批处理分析
    """

    def __init__(self):
        self.batch = BatchAnalyzer()
        self.pattern = PatternRecognizer()
        self._last_batch_time = 0

    def process_event(self, event: dict) -> tuple[RollingStats, list[MarketInsight]]:
        """
        处理单个事件, 返回实时统计和检测到的洞察
        """
        stats = self.batch.add_realtime_event(event)
        recent = list(self.batch.realtime._events)
        historical = []  # 可从存储加载历史数据
        insights = self.pattern.detect_anomaly([e for _, e in recent], stats)
        return stats, insights

    def run_batch_analysis(
        self,
        start_date: datetime | None = None,
        end_date: datetime | None = None,
        tenant_id: str = "default",
    ) -> BatchReport:
        """执行批处理分析"""
        if end_date is None:
            end_date = datetime.now(timezone.utc)
        if start_date is None:
            start_date = end_date - timedelta(hours=24)

        # 加载清洗后的数据
        events = self._load_cleaned_events(start_date, end_date, tenant_id)
        report = self.batch.analyze_period(events, start_date, end_date)
        return report

    def _load_cleaned_events(
        self,
        start: datetime,
        end: datetime,
        tenant_id: str,
    ) -> list[dict]:
        """从清洗后数据目录加载事件"""
        events = []
        current = start
        while current <= end:
            month_prefix = current.strftime("%Y-%m")
            month_dir = CLEANED_DATA_DIR / month_prefix
            if month_dir.exists():
                for f in month_dir.iterdir():
                    if f.suffix == ".ndjson":
                        with open(f, "r") as fp:
                            for line in fp:
                                try:
                                    e = json.loads(line.strip())
                                    ts_str = e.get("ts", "")
                                    if ts_str:
                                        for fmt in ("%Y-%m-%dT%H:%M:%S", "%Y-%m-%dT%H:%M:%SZ"):
                                            try:
                                                dt = datetime.strptime(ts_str[:19], fmt).replace(tzinfo=timezone.utc)
                                                if start <= dt <= end:
                                                    events.append(e)
                                                break
                                            except ValueError:
                                                continue
                                except json.JSONDecodeError:
                                    continue
            current += timedelta(days=1)
        return events

    def save_report(self, report: BatchReport, name: str = "daily") -> Path:
        """保存分析报告"""
        ANALYTICS_DIR.mkdir(parents=True, exist_ok=True)
        REPORTS_DIR.mkdir(parents=True, exist_ok=True)

        report_file = ANALYTICS_DIR / f"{name}_report_{datetime.now(timezone.utc).strftime('%Y%m%d_%H%M%S')}.json"
        report_file.write_text(json.dumps(asdict(report), ensure_ascii=False, indent=2), encoding="utf-8")

        return report_file


# ── 便捷函数 ─────────────────────────────────────────────────────

_engine: AnalyticsEngine | None = None


def get_engine() -> AnalyticsEngine:
    global _engine
    if _engine is None:
        _engine = AnalyticsEngine()
    return _engine


def compute_realtime_stats() -> RollingStats:
    return get_engine().batch.realtime.get_stats()


def run_daily_report(tenant_id: str = "default") -> BatchReport:
    engine = get_engine()
    end = datetime.now(timezone.utc)
    start = end - timedelta(hours=24)
    return engine.run_batch_analysis(start, end, tenant_id)


# ── 主入口 ───────────────────────────────────────────────────────

if __name__ == "__main__":
    import argparse

    parser = argparse.ArgumentParser(description="Analytics engine")
    parser.add_argument("--report", choices=["daily", "weekly"], default="daily")
    parser.add_argument("--tenant", default="default")
    args = parser.parse_args()

    engine = get_engine()
    report = engine.run_batch_analysis(tenant_id=args.tenant)
    print(json.dumps(asdict(report), ensure_ascii=False, indent=2))