#!/usr/bin/env python3
"""
data_cleaner.py — 数据清洗与预处理引擎

职责：
  1. 读取原始交易事件 (ndjson)
  2. 执行多级清洗：去重、空值填充、异常值剔除
  3. 标准化字段格式 (时间、金额、方向)
  4. 输出清洗后的结构化数据
  5. 生成清洗报告

处理流程：
  raw_event → dedup → schema_validate → normalize → outlier_filter → clean_event
"""
from __future__ import annotations

import hashlib
import json
import os
import re
import time
from collections import defaultdict
from dataclasses import dataclass, asdict, field
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Optional

from src.market_context import normalize_regime_label

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

APP_DIR = Path(__file__).resolve().parent.parent.parent
DATA_SERVER_ROOT = APP_DIR / "data" / "server"
CLEANED_DATA_DIR = APP_DIR / "data" / "consumed" / "cleaned"
STATS_DIR = APP_DIR / "logs"
DEDUP_WINDOW_SEC = 60  # 60秒内相同 ticket 视为重复

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

@dataclass
class CleanedTradeEvent:
    """标准化后的交易事件"""
    ts: str                          # ISO8601 UTC
    tenant_hash: str                 # 租户哈希 (10字符)
    sequence: int                    # 顺序号
    side: str                         # BUY / SELL
    symbol: str                       # XAUUSD
    volume: float                     # 手数 (3位有效数字)
    entry_price: float               # 入场价 (3位小数)
    exit_price: Optional[float]       # 出场价
    pnl: float                        # 盈亏 (美元)
    regime: str                       # canonical regime label
    confidence: float                 # 信号置信度 0-1
    execution_latency_ms: int         # 执行延迟
    order_type: str                  # MARKET / LIMIT / STOP
    duration_min: Optional[int]       # 持仓时长(分钟)
    spread_pips: Optional[float]     # 入场点差
    atr_value: Optional[float]        # ATR值
    risk_score: Optional[float]       # 风险评分 0-1
    cleaned_at: str = field(default_factory=lambda: datetime.now(timezone.utc).isoformat())


@dataclass
class CleaningReport:
    """清洗报告"""
    input_count: int = 0
    output_count: int = 0
    duplicates_removed: int = 0
    invalid_schema: int = 0
    outliers_removed: int = 0
    nulls_filled: int = 0
    duration_sec: float = 0.0
    errors: list[str] = field(default_factory=list)


# ── 辅助函数 ─────────────────────────────────────────────────────

def _hash_tenant(tenant_id: str) -> str:
    salt = os.environ.get("ANON_SALT", "LobsterQuant2026")
    return hashlib.sha256(f"{salt}{tenant_id}".encode()).hexdigest()[:10]


def _parse_ts(ts_str: str) -> datetime | None:
    """解析时间字符串为 datetime"""
    for fmt in ("%Y-%m-%dT%H:%M:%S", "%Y-%m-%dT%H:%M:%SZ", "%Y-%m-%d %H:%M:%S"):
        try:
            return datetime.strptime(ts_str[:19], fmt).replace(tzinfo=timezone.utc)
        except ValueError:
            continue
    return None


def _normalize_symbol(symbol: str) -> str:
    """标准化品种名称"""
    s = symbol.upper().strip()
    mapping = {"GOLD": "XAUUSD", "XAUUSDUSD": "XAUUSD", "XAUUSD.": "XAUUSD"}
    return mapping.get(s, s)


def _validate_side(side: str) -> bool:
    return side.upper() in ("BUY", "SELL", "LONG", "SHORT")


def _validate_regime(regime: str) -> bool:
    return normalize_regime_label(regime) in {
        "STRONG_TREND", "WEAK_TREND", "STEALTH_TREND",
        "RANGING", "TRANSITION", "VOLATILITY_SPIKE", "FAKE_TREND", "EXTREME", "UNKNOWN",
    }


# ── 去重器 ───────────────────────────────────────────────────────

class DedupCache:
    """基于时间窗口的去重缓存"""

    def __init__(self, window_sec: int = 60):
        self.window_sec = window_sec
        self._cache: dict[str, datetime] = {}  # event_key → timestamp

    def _make_key(self, event: dict) -> str:
        """生成事件唯一键: tenant+symbol+side+ts"""
        ts = event.get("ts", "")
        return f"{event.get('tenant_id', '')}:{event.get('symbol', '')}:{event.get('side', '')}:{ts}"

    def is_duplicate(self, event: dict) -> bool:
        key = self._make_key(event)
        now = datetime.now(timezone.utc)
        if key in self._cache:
            prev = self._cache[key]
            if (now - prev).total_seconds() < self.window_sec:
                return True
        self._cache[key] = now
        # 清理过期键
        self._cache = {
            k: v for k, v in self._cache.items()
            if (now - v).total_seconds() < self.window_sec * 2
        }
        return False


# ── 异常值检测器 ─────────────────────────────────────────────────

class OutlierDetector:
    """基于 IQR 的异常值检测"""

    def __init__(self, fields: list[str] = None):
        self.fields = fields or ["volume", "pnl", "spread_pips", "execution_latency_ms"]
        self._stats: dict[str, list[float]] = defaultdict(list)
        self._fitted = False

    def fit(self, events: list[dict]) -> None:
        """从样本学习统计分布"""
        for e in events:
            for f in self.fields:
                v = e.get(f)
                if v is not None and isinstance(v, (int, float)):
                    self._stats[f].append(float(v))
        self._fitted = True

    def _iqr_bounds(self, values: list[float], k: float = 1.5) -> tuple[float, float]:
        if len(values) < 4:
            return float("-inf"), float("inf")
        sorted_vals = sorted(values)
        n = len(sorted_vals)
        q1 = sorted_vals[int(n * 0.25)]
        q3 = sorted_vals[int(n * 0.75)]
        iqr = q3 - q1
        return q1 - k * iqr, q3 + k * iqr

    def is_outlier(self, event: dict, field: str, k: float = 1.5) -> bool:
        if not self._fitted or field not in self._stats:
            return False
        v = event.get(field)
        if v is None:
            return False
        lo, hi = self._iqr_bounds(self._stats[field], k)
        return v < lo or v > hi

    def is_extreme(self, event: dict) -> bool:
        """极端异常: 超过3倍IQR"""
        for f in self.fields:
            if self.is_outlier(event, f, k=3.0):
                return True
        return False


# ── 清洗引擎 ─────────────────────────────────────────────────────

class DataCleaner:
    """
    多级数据清洗引擎
    流程: dedup → schema_validate → normalize → outlier_filter
    """

    def __init__(self, fit_sample_size: int = 200):
        self.dedup = DedupCache(window_sec=DEDUP_WINDOW_SEC)
        self.outlier_detector = OutlierDetector()
        self.fit_sample_size = fit_sample_size
        self._dedup_fitted = False

    def _validate_schema(self, event: dict) -> bool:
        """检查必填字段"""
        required = ["ts", "side", "symbol", "volume"]
        for field in required:
            if field not in event or event[field] is None:
                return False
        return True

    def _normalize_event(self, event: dict) -> dict:
        """标准化字段格式"""
        normalized = event.copy()

        # 时间标准化
        if "ts" in normalized:
            dt = _parse_ts(normalized["ts"])
            if dt:
                normalized["ts"] = dt.strftime("%Y-%m-%dT%H:%M:%SZ")
            else:
                normalized["ts"] = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")

        # 品种标准化
        if "symbol" in normalized:
            normalized["symbol"] = _normalize_symbol(normalized["symbol"])

        # 方向标准化
        if "side" in normalized:
            s = normalized["side"].upper()
            normalized["side"] = "BUY" if s in ("BUY", "LONG") else "SELL"

        # regime 标准化
        if "regime" in normalized:
            normalized["regime"] = normalize_regime_label(normalized["regime"])

        # 数值字段标准化 (3位有效数字)
        for f in ("volume", "entry", "exit", "pnl"):
            if f in normalized and normalized[f] is not None:
                try:
                    normalized[f] = round(float(normalized[f]), 3)
                except (ValueError, TypeError):
                    pass

        # 空值填充
        for f in ("confidence", "execution_latency_ms", "spread_pips"):
            if f not in normalized or normalized[f] is None:
                defaults = {
                    "confidence": 0.5,
                    "execution_latency_ms": 500,
                    "spread_pips": 15.0,
                }
                normalized[f] = defaults.get(f)

        return normalized

    def fit(self, events: list[dict]) -> None:
        """从样本学习统计分布用于异常检测"""
        sample = events[: self.fit_sample_size]
        self.outlier_detector.fit(sample)

    def clean(self, event: dict) -> Optional[dict]:
        """
        清洗单个事件。
        返回清洗后的事件 dict，或 None (事件被过滤)。
        """
        # 阶段1: 去重
        if self.dedup.is_duplicate(event):
            return None

        # 阶段2: Schema 验证
        if not self._validate_schema(event):
            return None

        # 阶段3: 标准化
        normalized = self._normalize_event(event)

        # 阶段4: 异常值过滤 (软过滤，不丢弃极端异常，只标记)
        # 异常值写入但带标记
        is_extreme = self.outlier_detector.is_extreme(normalized)
        if is_extreme:
            normalized["_flagged"] = "extreme_outlier"

        return normalized

    def clean_batch(self, events: list[dict]) -> tuple[list[dict], CleaningReport]:
        """
        批量清洗事件。
        返回: (清洗后事件列表, 清洗报告)
        """
        report = CleaningReport()
        report.input_count = len(events)
        start = time.time()

        # 第一遍: 学习统计分布
        self.fit(events)

        # 第二遍: 执行清洗
        cleaned = []
        for event in events:
            try:
                result = self.clean(event)
                if result is None:
                    report.duplicates_removed += 1
                else:
                    cleaned.append(result)
            except Exception as e:
                report.errors.append(str(e))

        report.output_count = len(cleaned)
        report.duration_sec = round(time.time() - start, 3)
        return cleaned, report


# ── 存储写入 ─────────────────────────────────────────────────────

def write_cleaned_events(
    events: list[dict],
    tenant_id: str,
    output_dir: Path | None = None,
) -> Path:
    """
    将清洗后的事件追加写入 ndjson 文件。
    按年月分区: data/consumed/cleaned/<YYYY-MM>/<tenant_hash>.ndjson
    """
    if output_dir is None:
        output_dir = CLEANED_DATA_DIR

    dt = datetime.now(timezone.utc)
    month_prefix = dt.strftime("%Y-%m")

    tenant_hash = _hash_tenant(tenant_id)
    sub_dir = output_dir / month_prefix
    sub_dir.mkdir(parents=True, exist_ok=True)

    out_file = sub_dir / f"{tenant_hash}.ndjson"

    with open(out_file, "a", encoding="utf-8") as f:
        for event in events:
            # 确保清洗标记和元数据
            event["_cleaned"] = True
            event["_cleaned_at"] = datetime.now(timezone.utc).isoformat()
            f.write(json.dumps(event, ensure_ascii=False) + "\n")

    return out_file


def write_cleaning_report(report: CleaningReport, tenant_id: str) -> None:
    """写入清洗报告到日志"""
    report_file = STATS_DIR / f"cleaning_report_{tenant_id}.json"
    data = asdict(report)
    data["tenant_id"] = tenant_id
    data["generated_at"] = datetime.now(timezone.utc).isoformat()
    report_file.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8")


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

def clean_trades(
    input_file: Path,
    tenant_id: str,
    output_dir: Path | None = None,
) -> CleaningReport:
    """
    从原始交易文件清洗并输出清洗后数据。
    """
    if not input_file.exists():
        raise FileNotFoundError(f"Input file not found: {input_file}")

    events = []
    with open(input_file, "r", encoding="utf-8") as f:
        for line in f:
            line = line.strip()
            if not line:
                continue
            try:
                events.append(json.loads(line))
            except json.JSONDecodeError:
                continue

    cleaner = DataCleaner()
    cleaned_events, report = cleaner.clean_batch(events)

    if cleaned_events:
        write_cleaned_events(cleaned_events, tenant_id, output_dir)

    write_cleaning_report(report, tenant_id)

    return report


if __name__ == "__main__":
    import argparse

    parser = argparse.ArgumentParser(description="Data cleaner for trade events")
    parser.add_argument("--input", type=Path, required=True, help="Input ndjson file")
    parser.add_argument("--tenant", default="default", help="Tenant ID")
    parser.add_argument("--output-dir", type=Path, help="Output directory")
    args = parser.parse_args()

    report = clean_trades(args.input, args.tenant, args.output_dir)
    print(json.dumps(asdict(report), ensure_ascii=False, indent=2))
