"""
链上指标数据模块：拉取 BTC 交易所净流量、稳定币供应、ETF 净流入、MVRV Z-Score、SOPR。
本地 CSV 落盘 + 内存缓存，参照 derivatives 模式。

免费数据源：
  - CoinMetrics Community API: 交易所流入/流出(USD)、MVRV Ratio
  - CoinGecko: 稳定币市值
需要 API key 的指标（可配置）：
  - ETF 净流入: Coinglass / Farside / Glassnode
  - SOPR: Glassnode / CoinMetrics Pro
"""
import csv
import json
import logging
import os
import re
import threading
import time
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Dict, List, Optional

import requests
from fastapi import APIRouter, HTTPException, Query

logger = logging.getLogger("onchain_metrics")

# ── 常量 ─────────────────────────────────────────────────────────────
METRIC_NAMES = [
    "btc_netflow",
    "stablecoin_supply",
    "etf_flow",
    "mvrv_zscore",
    "sopr",
]
REFRESH_INTERVAL = 300  # 后台刷新间隔（秒），免费 API 限频较严
COINMETRICS_BASE = "https://community-api.coinmetrics.io/v4"
COINGECKO_BASE = "https://api.coingecko.com/api/v3"
REQUEST_TIMEOUT = 30
HTTP_HEADERS = {"User-Agent": "Mozilla/5.0 (compatible; chan-data-provider/1.0)"}


def to_utc_iso(ts_ms: int) -> str:
    dt = datetime.fromtimestamp(ts_ms / 1000, tz=timezone.utc)
    return dt.isoformat().replace("+00:00", "Z")


# ── OnchainMetricsManager ────────────────────────────────────────────
class OnchainMetricsManager:
    """管理链上指标的拉取、缓存和持久化。每个指标独立存储。"""

    def __init__(self, data_dir: Path, api_keys: Optional[Dict[str, str]] = None) -> None:
        self.data_dir = data_dir / "onchain"
        self.data_dir.mkdir(parents=True, exist_ok=True)
        self.api_keys = api_keys or {}

        # 内存缓存: metric_name -> List[Dict]
        self._metrics: Dict[str, List[Dict]] = {name: [] for name in METRIC_NAMES}
        self._lock = threading.RLock()
        self._stop_event = threading.Event()
        self._thread: Optional[threading.Thread] = None

        # 加载本地历史
        for name in METRIC_NAMES:
            local = self._load_local(name)
            if local:
                self._metrics[name] = local
                logger.info("链上指标 %s 加载本地记录: %d 条", name, len(local))

    # ── CSV 路径 ─────────────────────────────────────────────────────
    def _path(self, metric: str) -> Path:
        return self.data_dir / f"{metric}.csv"

    # ── CSV 读写 ─────────────────────────────────────────────────────
    def _load_local(self, metric: str) -> List[Dict]:
        path = self._path(metric)
        if not path.exists():
            return []
        records = []
        with path.open("r", encoding="utf-8", newline="") as fp:
            for row in csv.DictReader(fp):
                try:
                    records.append({
                        "timestamp": int(row["timestamp"]),
                        "datetime": row.get("datetime", ""),
                        "value": float(row.get("value", 0)),
                        "sub_value": float(row.get("sub_value", 0)) if row.get("sub_value") else None,
                        "extra": json.loads(row.get("extra", "{}")) if row.get("extra") else {},
                    })
                except (KeyError, ValueError):
                    continue
        records.sort(key=lambda r: r["timestamp"])
        return records

    def _write_local(self, metric: str, records: List[Dict]) -> None:
        path = self._path(metric)
        fieldnames = ["timestamp", "datetime", "value", "sub_value", "extra"]
        tmp = path.with_suffix(".tmp")
        try:
            with tmp.open("w", encoding="utf-8", newline="") as fp:
                w = csv.DictWriter(fp, fieldnames=fieldnames, extrasaction="ignore")
                w.writeheader()
                for r in records:
                    row = {
                        "timestamp": r.get("timestamp", 0),
                        "datetime": r.get("datetime", ""),
                        "value": r.get("value", 0),
                        "sub_value": "" if r.get("sub_value") is None else r["sub_value"],
                        "extra": json.dumps(r.get("extra", {}), ensure_ascii=False),
                    }
                    w.writerow(row)
            tmp.replace(path)
            logger.info("链上指标 %s 已写入磁盘: %d 条", metric, len(records))
        except Exception as e:
            logger.error("链上指标 %s 落盘失败: %s", metric, e)
            if tmp.exists():
                tmp.unlink()

    def _merge(self, base: List[Dict], new: List[Dict]) -> List[Dict]:
        """按 timestamp 去重合并，新覆盖旧。"""
        merged = {r["timestamp"]: r for r in base}
        for r in new:
            merged[r["timestamp"]] = r
        return sorted(merged.values(), key=lambda r: r["timestamp"])

    # ── 数据拉取 ─────────────────────────────────────────────────────
    def _fetch_cm_metrics(
        self, metrics: str, days: int = 365
    ) -> Optional[List[Dict]]:
        """从 CoinMetrics Community API 拉取指标。返回 [{time, metric: value}, ...]"""
        page_size = min(days, 10000)
        url = (
            f"{COINMETRICS_BASE}/timeseries/asset-metrics"
            f"?assets=btc&metrics={metrics}&frequency=1d&page_size={page_size}"
        )
        try:
            r = requests.get(url, timeout=REQUEST_TIMEOUT, headers=HTTP_HEADERS)
            if r.status_code != 200:
                logger.warning("CoinMetrics %s 返回 %d: %s", metrics, r.status_code, r.text[:200])
                return None
            data = r.json().get("data", [])
            if not data:
                return None
            # 翻页取更多历史
            all_data = list(data)
            next_url = r.json().get("next_page_url")
            pages = 0
            while next_url and pages < 10:
                pages += 1
                time.sleep(0.5)
                r2 = requests.get(next_url, timeout=REQUEST_TIMEOUT, headers=HTTP_HEADERS)
                if r2.status_code != 200:
                    break
                batch = r2.json()
                all_data.extend(batch.get("data", []))
                next_url = batch.get("next_page_url")
                if not next_url or next_url == r.json().get("next_page_url"):
                    break
            return all_data
        except Exception as e:
            logger.error("CoinMetrics %s 拉取失败: %s", metrics, e)
            return None

    def _fetch_btc_netflow(self) -> List[Dict]:
        """BTC 交易所净流量 (USD) = 流入 - 流出。"""
        data = self._fetch_cm_metrics("FlowInExUSD,FlowOutExUSD", days=365)
        if not data:
            return []

        records = []
        for d in data:
            ts_str = d.get("time", "")
            inflow = float(d.get("FlowInExUSD") or 0)
            outflow = float(d.get("FlowOutExUSD") or 0)
            netflow = inflow - outflow
            try:
                ts = int(datetime.fromisoformat(ts_str.replace("Z", "+00:00")).timestamp() * 1000)
            except Exception:
                continue
            records.append({
                "timestamp": ts,
                "datetime": to_utc_iso(ts),
                "value": round(netflow, 2),
                "sub_value": round(inflow, 2),
                "extra": {"inflow_usd": round(inflow, 2), "outflow_usd": round(outflow, 2)},
            })
        return sorted(records, key=lambda r: r["timestamp"])

    def _fetch_stablecoin_supply(self) -> List[Dict]:
        """稳定币总供应（USDT + USDC + DAI + FDUSD + TUSD 市值之和）。"""
        try:
            url = (
                f"{COINGECKO_BASE}/coins/markets"
                f"?vs_currency=usd&category=stablecoins&order=market_cap_desc"
                f"&per_page=5&page=1&sparkline=false"
            )
            r = requests.get(url, timeout=REQUEST_TIMEOUT, headers=HTTP_HEADERS)
            if r.status_code != 200:
                logger.warning("CoinGecko stablecoins 返回 %d", r.status_code)
                return []
            coins = r.json()
            if not isinstance(coins, list):
                return []

            total_mcap = sum(c.get("market_cap", 0) or 0 for c in coins)
            now_ms = int(time.time() * 1000)
            breakdown = {
                c.get("symbol", "?").upper(): c.get("market_cap", 0) or 0
                for c in coins[:5]
            }
            return [{
                "timestamp": now_ms,
                "datetime": to_utc_iso(now_ms),
                "value": round(total_mcap, 2),
                "sub_value": round(total_mcap / 1e9, 2),  # Billions
                "extra": breakdown,
            }]
        except Exception as e:
            logger.error("稳定币供应拉取失败: %s", e)
            return []

    def _fetch_etf_flow(self) -> List[Dict]:
        """BTC ETF 净流入/流出。优先从 Farside 免费爬取，Coinglass 为备用。

        数据源优先级:
          1. Farside (免费, HTML 爬取)
          2. Coinglass (需 coinglass_key)
          3. Glassnode (需 glassnode_key)
        """
        # ── 优先：Farside 免费爬取 ──
        try:
            records = self._fetch_etf_farside()
            if records:
                logger.info("ETF 数据从 Farside 拉取: %d 条", len(records))
                return records
        except Exception as e:
            logger.warning("Farside ETF 爬取失败: %s", e)

        # ── 备用：Coinglass ──
        cg_key = self.api_keys.get("coinglass_key")
        if cg_key:
            try:
                url = "https://open-api-v3.coinglass.com/api/bitcoin/etf/net-inflow?interval=30"
                r = requests.get(url, timeout=REQUEST_TIMEOUT, headers={
                    **HTTP_HEADERS, "coinglassSecret": cg_key,
                })
                if r.status_code == 200:
                    data = r.json()
                    records = []
                    if isinstance(data, dict) and "data" in data:
                        for item in data["data"]:
                            ts = int(item.get("date", 0)) * 1000 if item.get("date") else 0
                            if ts:
                                records.append({
                                    "timestamp": ts,
                                    "datetime": to_utc_iso(ts),
                                    "value": float(item.get("netInflow", 0)),
                                    "sub_value": None,
                                    "extra": item,
                                })
                    return records
                logger.warning("Coinglass ETF 返回 %d", r.status_code)
            except Exception as e:
                logger.error("Coinglass ETF 拉取失败: %s", e)

        logger.warning("ETF 数据源均不可用（Farside/Coinglass/Glassnode）")
        return []

    def _fetch_etf_farside(self) -> List[Dict]:
        """从 Farside 网站爬取 BTC ETF 每日净流量。

        https://farside.co.uk/btc/ 页面包含一个 HTML 表格，
        每行包含日期和各 ETF 的当日流量（百万美元），最后一列为总计。
        """
        url = "https://farside.co.uk/btc/"
        r = requests.get(url, timeout=REQUEST_TIMEOUT, headers={
            **HTTP_HEADERS,
            "Accept": "text/html,application/xhtml+xml,*/*",
        })
        if r.status_code != 200:
            logger.warning("Farside 返回 %d", r.status_code)
            return []

        html = r.text

        # 查找表格
        table_match = re.search(r'<table[^>]*>(.*?)</table>', html, re.DOTALL)
        if not table_match:
            logger.warning("Farside 页面未找到表格")
            return []

        rows_html = re.findall(r'<tr[^>]*>(.*?)</tr>', table_match.group(1), re.DOTALL)
        month_map = {
            "jan": 1, "feb": 2, "mar": 3, "apr": 4, "may": 5, "jun": 6,
            "jul": 7, "aug": 8, "sep": 9, "oct": 10, "nov": 11, "dec": 12,
        }

        records = []
        for row_html in rows_html:
            cells = re.findall(r'<t[dh][^>]*>(.*?)</t[dh]>', row_html, re.DOTALL)
            # 清理 HTML 标签和空白
            clean = []
            for c in cells:
                t = re.sub(r'<[^>]+>', '', c).strip()
                t = t.replace('\xa0', ' ').replace('&nbsp;', ' ').strip()
                clean.append(t)

            if not clean:
                continue

            # 第一列应为日期格式 "15 Jun 2026"
            date_str = clean[0]
            parts = date_str.split()
            if len(parts) != 3:
                continue
            day_str, mon_str, year_str = parts
            mon = month_map.get(mon_str.lower()[:3])
            if mon is None:
                continue
            try:
                day = int(day_str)
                year = int(year_str)
            except ValueError:
                continue

            # 最后一列为总计（可能带括号表示负值）
            total_str = clean[-1] if len(clean) > 1 else ""
            if not total_str or total_str in ("", "Total", "-"):
                continue
            # 解析 "(123.4)" → -123.4, "123.4" → 123.4
            total_str = total_str.replace(",", "")
            is_negative = total_str.startswith("(") and total_str.endswith(")")
            if is_negative:
                total_str = total_str[1:-1]
            try:
                total_m = float(total_str)
            except ValueError:
                continue
            if is_negative:
                total_m = -total_m

            # 构建时间戳（UTC 午夜）
            from datetime import datetime, timezone as tz
            dt = datetime(year, mon, day, tzinfo=tz.utc)
            ts = int(dt.timestamp() * 1000)

            # 分解各 ETF 明细
            etf_breakdown = {}
            if len(clean) > 2:
                etf_names = ["IBIT", "FBTC", "BITB", "ARKB", "BTCO", "EZBC",
                             "BRRR", "HODL", "BTCW", "MSBT", "GBTC", "BTC"]
                for i, name in enumerate(etf_names):
                    idx = i + 1
                    if idx < len(clean) - 1:
                        val_str = clean[idx].replace(",", "").replace("(", "").replace(")", "")
                        neg = clean[idx].startswith("(")
                        try:
                            val = float(val_str)
                            etf_breakdown[name] = -val if neg else val
                        except ValueError:
                            pass

            records.append({
                "timestamp": ts,
                "datetime": dt.isoformat().replace("+00:00", "Z"),
                "value": round(total_m, 2),      # 净流入 (百万 USD)
                "sub_value": round(total_m * 1_000_000, 2),  # 净流入 (USD)
                "extra": {"total_million_usd": round(total_m, 2), "breakdown": etf_breakdown},
            })

        return sorted(records, key=lambda r: r["timestamp"])

    def _fetch_mvrv_zscore(self) -> List[Dict]:
        """MVRV Z-Score = (当前 MVRV - 滚动均值) / 滚动标准差。

        从 CoinMetrics 拉取 CapMVRVCur（免费），计算 365 天滚动 Z-Score。
        """
        data = self._fetch_cm_metrics("CapMVRVCur", days=400)
        if not data:
            return []

        # 解析时间序列
        mvrv_series = []
        for d in data:
            ts_str = d.get("time", "")
            mvrv_val = d.get("CapMVRVCur")
            if mvrv_val is None:
                continue
            try:
                ts = int(datetime.fromisoformat(ts_str.replace("Z", "+00:00")).timestamp() * 1000)
            except Exception:
                continue
            mvrv_series.append((ts, float(mvrv_val)))
        mvrv_series.sort(key=lambda x: x[0])

        # 计算滚动 Z-Score (365 天窗口，即约 365 个数据点)
        window = 365
        records = []
        values = []
        timestamps = []
        for ts, val in mvrv_series:
            values.append(val)
            timestamps.append(ts)
            if len(values) < window:
                continue
            window_vals = values[-window:]
            mean = sum(window_vals) / window
            variance = sum((v - mean) ** 2 for v in window_vals) / window
            stddev = variance ** 0.5
            zscore = (val - mean) / stddev if stddev > 0 else 0.0
            records.append({
                "timestamp": ts,
                "datetime": to_utc_iso(ts),
                "value": round(zscore, 4),
                "sub_value": round(val, 4),
                "extra": {"mvrv_ratio": round(val, 4), "rolling_mean": round(mean, 4), "rolling_stddev": round(stddev, 4)},
            })
        return records

    def _fetch_sopr(self) -> List[Dict]:
        """SOPR (Spent Output Profit Ratio)。需要 API key。

        支持:
          - glassnode_key: Glassnode v1/metrics/indicators/sopr
        """
        gn_key = self.api_keys.get("glassnode_key")
        if gn_key:
            try:
                url = (
                    f"https://api.glassnode.com/v1/metrics/indicators/sopr"
                    f"?a=btc&f=json&api_key={gn_key}"
                )
                r = requests.get(url, timeout=REQUEST_TIMEOUT, headers=HTTP_HEADERS)
                if r.status_code == 200:
                    data = r.json()
                    records = []
                    for item in data if isinstance(data, list) else []:
                        ts = int(item.get("t", 0)) * 1000
                        if ts:
                            records.append({
                                "timestamp": ts,
                                "datetime": to_utc_iso(ts),
                                "value": float(item.get("v", 0)),
                                "sub_value": None,
                                "extra": {"raw": item},
                            })
                    return records
                logger.warning("Glassnode SOPR 返回 %d", r.status_code)
            except Exception as e:
                logger.error("Glassnode SOPR 拉取失败: %s", e)

        logger.warning(
            "SOPR 数据未配置 API key。请设置环境变量 GLASSNODE_API_KEY"
        )
        return []

    # ── 刷新入口 ─────────────────────────────────────────────────────
    def refresh_all(self) -> None:
        """拉取全部 5 个指标，合并到内存并写盘。"""
        fetchers = {
            "btc_netflow": self._fetch_btc_netflow,
            "stablecoin_supply": self._fetch_stablecoin_supply,
            "etf_flow": self._fetch_etf_flow,
            "mvrv_zscore": self._fetch_mvrv_zscore,
            "sopr": self._fetch_sopr,
        }

        for name, fetcher in fetchers.items():
            try:
                new_records = fetcher()
                if not new_records:
                    continue
                with self._lock:
                    base = list(self._metrics.get(name, []))
                    merged = self._merge(base, new_records)
                    self._metrics[name] = merged
                self._write_local(name, merged)
                logger.info("链上指标 %s 刷新: +%d 条, 总计 %d 条",
                            name, len(new_records), len(merged))
            except Exception:
                logger.exception("链上指标 %s 刷新异常", name)

    # ── 后台线程 ─────────────────────────────────────────────────────
    def _refresh_loop(self) -> None:
        """后台线程：启动立即拉取一次，之后每 REFRESH_INTERVAL 秒刷新。"""
        logger.info("链上指标后台线程启动，间隔 %ds", REFRESH_INTERVAL)
        # 启动立即拉取
        self.refresh_all()
        while not self._stop_event.wait(REFRESH_INTERVAL):
            self.refresh_all()
        logger.info("链上指标后台线程已退出")

    def start(self) -> None:
        self._stop_event.clear()
        self._thread = threading.Thread(
            target=self._refresh_loop, name="onchain-loop", daemon=True,
        )
        self._thread.start()
        logger.info("链上指标模块已启动")

    def stop(self) -> None:
        self._stop_event.set()
        if self._thread:
            self._thread.join(timeout=5)

    # ── 查询接口 ─────────────────────────────────────────────────────
    def get_metric(
        self,
        metric: str,
        limit: int = 100,
        start_ms: Optional[int] = None,
        end_ms: Optional[int] = None,
    ) -> List[Dict]:
        if metric not in METRIC_NAMES:
            raise HTTPException(
                status_code=400,
                detail=f"不支持的指标: {metric}。可选: {METRIC_NAMES}",
            )
        with self._lock:
            records = list(self._metrics.get(metric, []))
        if start_ms is not None:
            records = [r for r in records if r["timestamp"] >= start_ms]
        if end_ms is not None:
            records = [r for r in records if r["timestamp"] <= end_ms]
        return records[-limit:] if limit else records

    def get_latest(self) -> Dict[str, Any]:
        """获取所有指标的最新值。"""
        result = {"timestamp": int(time.time() * 1000)}
        for name in METRIC_NAMES:
            with self._lock:
                records = self._metrics.get(name, [])
            if records:
                latest = records[-1]
                result[name] = {
                    "timestamp": latest["timestamp"],
                    "datetime": latest["datetime"],
                    "value": latest["value"],
                    "sub_value": latest.get("sub_value"),
                    "extra": latest.get("extra", {}),
                }
            else:
                result[name] = None
        return result


# ── FastAPI Router ────────────────────────────────────────────────────
def create_onchain_router(manager: OnchainMetricsManager) -> APIRouter:
    router = APIRouter(prefix="/api/onchain", tags=["onchain"])

    @router.get("/metrics")
    async def get_metrics(
        metric: str = Query(..., description=f"指标名称: {', '.join(METRIC_NAMES)}"),
        limit: int = Query(100, ge=1, le=5000, description="返回条数上限"),
        start: Optional[int] = Query(None, description="起始时间戳(ms)"),
        end: Optional[int] = Query(None, description="结束时间戳(ms)"),
    ):
        """获取单个链上指标的时间序列。"""
        data = manager.get_metric(metric, limit=limit, start_ms=start, end_ms=end)
        return {"metric": metric, "count": len(data), "data": data}

    @router.get("/latest")
    async def get_latest():
        """获取全部 5 个指标的最新快照。"""
        return manager.get_latest()

    @router.get("/available")
    async def get_available():
        """列出可用指标及当前数据量。"""
        info = {}
        for name in METRIC_NAMES:
            records = manager.get_metric(name, limit=0)
            info[name] = {
                "count": len(records),
                "has_data": len(records) > 0,
                "first_ts": records[0]["datetime"] if records else None,
                "last_ts": records[-1]["datetime"] if records else None,
            }
        return info

    return router


# ── 环境变量辅助 ─────────────────────────────────────────────────────
def api_keys_from_env() -> Dict[str, str]:
    """从环境变量读取 API keys。"""
    keys = {}
    for env_var, key_name in [
        ("COINGLASS_API_KEY", "coinglass_key"),
        ("GLASSNODE_API_KEY", "glassnode_key"),
        ("FARSIDE_API_KEY", "farside_key"),
        ("COINMETRICS_API_KEY", "coinmetrics_key"),
    ]:
        val = os.getenv(env_var, "").strip()
        if val and "***" not in val:  # 忽略占位符
            keys[key_name] = val
    return keys