"""研究用数据层：本地历史数据优先，回落到远端数据服务。

远端服务的小周期只保留 90 天（派生自 1m），做区间套时样本严重不足。
本项目 data/ 下是完整下载的 BTC/ETH/SOL 全周期数据（1m~1w，2172~2543 天），
是主力数据源；旧的 freqtrade 目录作为备用。
"""
from __future__ import annotations

from pathlib import Path

import pandas as pd
import requests

DATA_SERVICE_URL = "https://provider.jackyu66.com"
CACHE_DIR = Path(__file__).resolve().parents[1] / ".cache"
NUMERIC_COLS = ("open", "high", "low", "close", "volume")

# 按优先级查找，第一个命中的目录生效
LOCAL_DATA_DIRS = (
    Path(__file__).resolve().parents[2] / "data",
    Path("/Users/jack/Project/freqtrade/user_data/data"),
)


def _cache_path(symbol: str, tf: str, limit: int) -> Path:
    safe = symbol.replace("/", "_").replace(":", "-")
    return CACHE_DIR / f"{safe}__{tf}__{limit}.parquet"


def _normalize(raw: list[dict]) -> pd.DataFrame:
    df = pd.DataFrame(raw)
    df["timestamp"] = pd.to_numeric(df["timestamp"], errors="coerce")
    for col in NUMERIC_COLS:
        df[col] = pd.to_numeric(df[col], errors="coerce")
    df = df.dropna(subset=["timestamp", *NUMERIC_COLS])
    df = df.drop_duplicates(subset=["timestamp"]).sort_values("timestamp").reset_index(drop=True)
    df["timestamp"] = df["timestamp"].astype("int64")
    df["date"] = pd.to_datetime(df["timestamp"], unit="ms", utc=True).dt.tz_convert("Asia/Shanghai")
    return df[["timestamp", "date", *NUMERIC_COLS]]


def _freqtrade_path(symbol: str, tf: str, exchange: str = "binance") -> Path | None:
    """定位 feather 文件。BTC/USDT:USDT -> BTC_USDT_USDT-1h-futures.feather"""
    for root in LOCAL_DATA_DIRS:
        base = root / exchange
        if ":" in symbol:  # 永续合约
            name = symbol.replace("/", "_").replace(":", "_")
            cand = base / "futures" / f"{name}-{tf}-futures.feather"
        else:
            name = symbol.replace("/", "_")
            cand = base / f"{name}-{tf}.feather"
        if cand.exists():
            return cand
    return None


def load_local(symbol: str, tf: str, exchange: str = "binance") -> pd.DataFrame | None:
    """读取 freqtrade 本地历史数据，转成与远端一致的列结构。"""
    path = _freqtrade_path(symbol, tf, exchange)
    if path is None:
        return None
    df = pd.read_feather(path)
    df = df.rename(columns={c: c.lower() for c in df.columns})
    if "date" not in df.columns:
        return None
    date = pd.to_datetime(df["date"], utc=True)
    # 不能用 date.astype("int64")//10**6：该值的单位取决于列的精度
    # （datetime64[ns] 给纳秒、[ms] 给毫秒），对毫秒精度的文件会把时间戳砸平，
    # 进而被 drop_duplicates 删掉九成数据。下面的写法与底层精度无关。
    epoch_ms = (date - pd.Timestamp("1970-01-01", tz="UTC")) // pd.Timedelta("1ms")
    out = pd.DataFrame({
        "timestamp": epoch_ms.astype("int64"),
        "date": date.dt.tz_convert("Asia/Shanghai"),
    })
    for col in NUMERIC_COLS:
        out[col] = pd.to_numeric(df[col], errors="coerce") if col in df else 0.0
    out = out.dropna(subset=list(NUMERIC_COLS))
    return out.drop_duplicates(subset=["timestamp"]).sort_values("timestamp").reset_index(drop=True)


def fetch_ohlcv(
    symbol: str,
    tf: str,
    limit: int = 5000,
    refresh: bool = False,
    prefer_local: bool = True,
) -> pd.DataFrame:
    """获取K线。优先本地 freqtrade 数据，其次本地缓存，最后回源数据服务。"""
    if prefer_local and not refresh:
        local = load_local(symbol, tf)
        if local is not None and len(local) > 0:
            return local.tail(limit).reset_index(drop=True) if limit and len(local) > limit else local

    path = _cache_path(symbol, tf, limit)
    if path.exists() and not refresh:
        return pd.read_parquet(path)

    resp = requests.get(
        f"{DATA_SERVICE_URL}/api/candles",
        params={"symbol": symbol, "tf": tf, "limit": limit},
        timeout=60,
    )
    resp.raise_for_status()
    df = _normalize(resp.json())

    CACHE_DIR.mkdir(parents=True, exist_ok=True)
    df.to_parquet(path, index=False)
    return df