#!/usr/bin/env python3
"""
quality_monitor.py — 数据质量监控引擎

职责:
  1. 完整性检查: 必填字段缺失率
  2. 一致性检查: 字段类型、范围、格式校验
  3. 时效性检查: 数据新鲜度、数据链路延迟
  4. 唯一性检查: 重复事件检测
  5. 异常告警: 质量问题自动告警

监控指标:
  - completeness: 必填字段完整率 (目标 ≥95%)
  - consistency: 数据格式正确率 (目标 ≥98%)
  - freshness: 数据新鲜度 (目标 ≤5min)
  - uniqueness: 去重率 (目标 ≥90%)
  - volume: 数据量异常检测
"""
from __future__ import annotations

import json
import os
import re
import time
from collections import defaultdict
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
DATA_SERVER_ROOT = APP_DIR / "data" / "server"
CLEANED_DATA_DIR = APP_DIR / "data" / "consumed" / "cleaned"
QUALITY_REPORTS_DIR = APP_DIR / "reports" / "quality"
QUALITY_LOG = APP_DIR / "logs" / "quality_monitor.ndjson"

# 告警阈值
THRESHOLDS = {
    "completeness_pct": 95.0,   # 必填字段完整率 < 95% → WARN
    "consistency_pct": 98.0,    # 格式正确率 < 98% → WARN
    "freshness_sec": 300.0,      # 数据延迟 > 5min → WARN
    "duplicate_pct": 10.0,       # 重复率 > 10% → ERROR
    "volume_anomaly_pct": 50.0,  # 数据量比预期波动 > 50% → WARN
}


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

@dataclass
class FieldCheck:
    """字段检查结果"""
    field: str
    present_count: int = 0
    missing_count: int = 0
    null_count: int = 0
    completeness_pct: float = 0.0


@dataclass
class ConsistencyCheck:
    """一致性检查结果"""
    field: str
    valid_count: int = 0
    invalid_count: int = 0
    consistency_pct: float = 0.0
    invalid_examples: list[str] = field(default_factory=list)


@dataclass
class FreshnessCheck:
    """时效性检查结果"""
    latest_event_ts: str = ""
    age_seconds: float = 0.0
    freshness_pct: float = 100.0  # 0-100, 100=最新
    is_stale: bool = False


@dataclass
class VolumeCheck:
    """数据量检查结果"""
    event_count: int = 0
    expected_count: int = 0
    deviation_pct: float = 0.0
    is_anomaly: bool = False


@dataclass
class QualityReport:
    """完整质量报告"""
    tenant_id: str = "default"
    check_ts: str = ""
    period_start: str = ""
    period_end: str = ""

    total_events: int = 0
    field_checks: list[FieldCheck] = field(default_factory=list)
    consistency_checks: list[ConsistencyCheck] = field(default_factory=list)
    freshness_check: FreshnessCheck | None = None
    volume_check: VolumeCheck | None = None
    duplicate_count: int = 0
    duplicate_pct: float = 0.0

    overall_score: float = 0.0  # 0-100
    status: str = "OK"           # OK / WARN / ERROR
    alerts: list[str] = field(default_factory=list)
    recommendations: list[str] = field(default_factory=list)


# ── 必填字段定义 ─────────────────────────────────────────────────

REQUIRED_FIELDS = ["ts", "side", "symbol", "volume", "pnl", "regime"]

FIELD_SCHEMAS = {
    "ts": {"type": "string", "pattern": r"^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}"},
    "side": {"type": "string", "enum": ["BUY", "SELL", "LONG", "SHORT"]},
    "symbol": {"type": "string", "pattern": r"^[A-Z0-9]+$"},
    "volume": {"type": "number", "min": 0.001, "max": 100},
    "pnl": {"type": "number", "min": -10000, "max": 10000},
    "regime": {"type": "string", "enum": ["TRENDING", "RANGING", "VOLATILE"]},
    "confidence": {"type": "number", "min": 0, "max": 1},
    "execution_latency_ms": {"type": "number", "min": 0, "max": 60000},
}


# ── 完整性检查 ──────────────────────────────────────────────────

def check_completeness(events: list[dict]) -> list[FieldCheck]:
    """检查必填字段完整性"""
    checks = []
    total = len(events)
    if total == 0:
        return checks

    for field in REQUIRED_FIELDS:
        fc = FieldCheck(field=field)
        present = sum(1 for e in events if field in e and e[field] is not None)
        fc.present_count = present
        fc.missing_count = total - present
        fc.null_count = sum(1 for e in events if field in e and e[field] is None)
        fc.completeness_pct = round(present / total * 100, 2)
        checks.append(fc)

    return checks


# ── 一致性检查 ──────────────────────────────────────────────────

def _validate_value(field: str, value: Any) -> bool:
    """验证单个字段值"""
    schema = FIELD_SCHEMAS.get(field)
    if not schema:
        return True

    if schema.get("type") == "string":
        if "enum" in schema:
            return value in schema["enum"]
        if "pattern" in schema:
            return bool(re.match(schema["pattern"], str(value)))
    elif schema.get("type") == "number":
        try:
            v = float(value)
            if "min" in schema and v < schema["min"]:
                return False
            if "max" in schema and v > schema["max"]:
                return False
            return True
        except (ValueError, TypeError):
            return False

    return True


def check_consistency(events: list[dict]) -> list[ConsistencyCheck]:
    """检查数据格式一致性"""
    checks = []
    total = len(events)
    if total == 0:
        return checks

    for field, schema in FIELD_SCHEMAS.items():
        cc = ConsistencyCheck(field=field)
        valid = 0
        invalid_examples = []

        for e in events:
            if field not in e:
                continue
            if _validate_value(field, e[field]):
                valid += 1
            else:
                cc.invalid_count += 1
                if len(invalid_examples) < 3:
                    invalid_examples.append(f"{field}={e[field]}")

        cc.valid_count = valid
        cc.invalid_count = total - valid
        cc.consistency_pct = round(valid / total * 100, 2) if total > 0 else 100.0
        cc.invalid_examples = invalid_examples
        checks.append(cc)

    return checks


# ── 时效性检查 ──────────────────────────────────────────────────

def check_freshness(events: list[dict]) -> FreshnessCheck:
    """检查数据新鲜度"""
    fc = FreshnessCheck()
    if not events:
        fc.is_stale = True
        return fc

    # 找最新事件时间戳
    latest_ts = ""
    for e in events:
        ts = e.get("ts", "")
        if ts and ts > latest_ts:
            latest_ts = ts

    if not latest_ts:
        fc.is_stale = True
        return fc

    fc.latest_event_ts = latest_ts

    # 计算时间差
    for fmt in ("%Y-%m-%dT%H:%M:%S", "%Y-%m-%dT%H:%M:%SZ"):
        try:
            dt = datetime.strptime(latest_ts[:19], fmt).replace(tzinfo=timezone.utc)
            fc.age_seconds = (datetime.now(timezone.utc) - dt).total_seconds()
            fc.freshness_pct = max(0, 100 - fc.age_seconds / 30)  # 30秒内=100%
            fc.is_stale = fc.age_seconds > THRESHOLDS["freshness_sec"]
            break
        except ValueError:
            continue

    return fc


# ── 数据量检查 ──────────────────────────────────────────────────

def check_volume(events: list[dict], expected_per_hour: int = 10) -> VolumeCheck:
    """检查数据量异常"""
    vc = VolumeCheck()
    vc.event_count = len(events)
    vc.expected_count = expected_per_hour

    if expected_per_hour > 0:
        vc.deviation_pct = abs(vc.event_count - expected_per_hour) / expected_per_hour * 100
        vc.is_anomaly = vc.deviation_pct > THRESHOLDS["volume_anomaly_pct"]

    return vc


# ── 去重检查 ────────────────────────────────────────────────────

def check_uniqueness(events: list[dict]) -> tuple[int, float]:
    """检查重复率"""
    if not events:
        return 0, 0.0

    seen = set()
    duplicates = 0

    for e in events:
        # 生成唯一键: tenant + symbol + side + ts + volume
        key = f"{e.get('tenant_id', '')}:{e.get('symbol', '')}:{e.get('side', '')}:{e.get('ts', '')}:{e.get('volume', '')}"
        if key in seen:
            duplicates += 1
        else:
            seen.add(key)

    dup_pct = round(duplicates / len(events) * 100, 2)
    return duplicates, dup_pct


# ── 告警生成 ─────────────────────────────────────────────────────

def generate_alerts(report: QualityReport) -> list[str]:
    """根据质量报告生成告警"""
    alerts = []

    # 完整性告警
    for fc in report.field_checks:
        if fc.completeness_pct < THRESHOLDS["completeness_pct"]:
            alerts.append(f"字段 '{fc.field}' 完整率 {fc.completeness_pct}% < {THRESHOLDS['completeness_pct']}%")

    # 一致性告警
    for cc in report.consistency_checks:
        if cc.consistency_pct < THRESHOLDS["consistency_pct"]:
            alerts.append(f"字段 '{cc.field}' 格式正确率 {cc.consistency_pct}% < {THRESHOLDS['consistency_pct']}%")

    # 时效性告警
    if report.freshness_check and report.freshness_check.is_stale:
        age = report.freshness_check.age_seconds
        alerts.append(f"数据陈旧: 最新事件 {age:.0f}秒前, 超过 {THRESHOLDS['freshness_sec']}秒阈值")

    # 重复告警
    if report.duplicate_pct > THRESHOLDS["duplicate_pct"]:
        alerts.append(f"重复率 {report.duplicate_pct}% 超过 {THRESHOLDS['duplicate_pct']}% 阈值")

    return alerts


def generate_recommendations(report: QualityReport) -> list[str]:
    """根据质量报告生成改进建议"""
    recommendations = []

    for fc in report.field_checks:
        if fc.completeness_pct < 90:
            recommendations.append(f"提高 '{fc.field}' 字段采集率，当前 {fc.completeness_pct}%")

    for cc in report.consistency_checks:
        if cc.invalid_count > 0:
            recommendations.append(f"修复 '{cc.field}' 字段格式，当前 {cc.invalid_count} 条错误")

    if report.freshness_check and report.freshness_check.is_stale:
        recommendations.append("检查数据上传链路，降低延迟")

    if report.duplicate_pct > 5:
        recommendations.append("优化去重逻辑，减少重复事件")

    return recommendations


# ── 综合质量报告生成 ────────────────────────────────────────────

def _load_recent_events(tenant_id: str, hours: int = 24) -> list[dict]:
    """加载最近N小时的清洗后事件"""
    events = []
    cutoff = datetime.now(timezone.utc) - timedelta(hours=hours)

    # 扫描清洗后数据目录
    for month_dir in CLEANED_DATA_DIR.glob("????-??"):
        if not month_dir.is_dir():
            continue
        for f in month_dir.glob("*.ndjson"):
            try:
                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 dt >= cutoff:
                                            events.append(e)
                                        break
                                    except ValueError:
                                        continue
                        except json.JSONDecodeError:
                            continue
            except Exception:
                continue

    return events


def generate_quality_report(tenant_id: str = "default", hours: int = 24) -> dict[str, Any]:
    """
    生成完整的数据质量报告
    """
    report = QualityReport()
    report.tenant_id = tenant_id
    report.check_ts = datetime.now(timezone.utc).isoformat()
    report.period_start = (datetime.now(timezone.utc) - timedelta(hours=hours)).isoformat()
    report.period_end = datetime.now(timezone.utc).isoformat()

    events = _load_recent_events(tenant_id, hours)
    report.total_events = len(events)

    if events:
        # 完整性检查
        report.field_checks = check_completeness(events)
        # 一致性检查
        report.consistency_checks = check_consistency(events)
        # 时效性检查
        report.freshness_check = check_freshness(events)
        # 数据量检查
        report.volume_check = check_volume(events)
        # 去重检查
        report.duplicate_count, report.duplicate_pct = check_uniqueness(events)

        # 计算整体评分
        completeness_scores = [fc.completeness_pct for fc in report.field_checks]
        consistency_scores = [cc.consistency_pct for cc in report.consistency_checks]
        freshness_score = report.freshness_check.freshness_pct if report.freshness_check else 0

        if completeness_scores and consistency_scores:
            overall = (
                sum(completeness_scores) / len(completeness_scores) * 0.3 +
                sum(consistency_scores) / len(consistency_scores) * 0.3 +
                freshness_score * 0.2 +
                (100 - report.duplicate_pct) * 0.2
            )
            report.overall_score = round(overall, 1)

        # 告警和建议
        report.alerts = generate_alerts(report)
        report.recommendations = generate_recommendations(report)

        # 状态判定
        if report.alerts:
            error_count = sum(1 for a in report.alerts if "ERROR" in a or report.duplicate_pct > 20)
            report.status = "ERROR" if error_count > 0 else "WARN"
        else:
            report.status = "OK"

    return asdict(report)


def save_quality_report(report: dict[str, Any], tenant_id: str = "default") -> Path:
    """保存质量报告到文件"""
    QUALITY_REPORTS_DIR.mkdir(parents=True, exist_ok=True)
    ts = datetime.now(timezone.utc).strftime("%Y%m%d_%H%M%S")
    path = QUALITY_REPORTS_DIR / f"quality_report_{tenant_id}_{ts}.json"
    path.write_text(json.dumps(report, ensure_ascii=False, indent=2), encoding="utf-8")
    return path


def log_quality_event(report: dict[str, Any]) -> None:
    """追加质量事件到日志"""
    if QUALITY_LOG.parent.exists():
        with open(QUALITY_LOG, "a", encoding="utf-8") as f:
            f.write(json.dumps(report, ensure_ascii=False) + "\n")


def check_and_alert_quality() -> list[str]:
    """
    执行质量检查并返回告警列表 (供 autonomy_loop 调用)
    """
    report = generate_quality_report()
    save_quality_report(report)
    log_quality_event(report)

    if report["alerts"]:
        return report["alerts"]
    return []


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

def get_quality_score(tenant_id: str = "default") -> float:
    """获取当前质量评分"""
    report = generate_quality_report(tenant_id)
    return report.get("overall_score", 0.0)


def get_quality_status(tenant_id: str = "default") -> str:
    """获取当前质量状态"""
    report = generate_quality_report(tenant_id)
    return report.get("status", "UNKNOWN")


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

if __name__ == "__main__":
    import argparse

    parser = argparse.ArgumentParser(description="Data quality monitor")
    parser.add_argument("--tenant", default="default")
    parser.add_argument("--hours", type=int, default=24)
    parser.add_argument("--save", action="store_true", help="Save report to file")
    args = parser.parse_args()

    report = generate_quality_report(args.tenant, args.hours)
    if args.save:
        path = save_quality_report(report, args.tenant)
        print(f"Saved: {path}")

    print(json.dumps(report, ensure_ascii=False, indent=2))