"""Qlib 数据落盘 Provider（qlib_adapter 内部，业务层不直接 import）。

把本地行情长表（来自 SQLite，见 quant.service.bars_to_daily_df）按 qlib 0.9.8
的二进制格式导出到指定目录（provider_uri），供 qlib D.features 读取。

格式要点（来自实测 qlib 0.9.8.dev32 与源码 file_storage）：
- instruments 文件 3 列：instrument\\tstart\\tend（instrument 全小写）
- feature bin 布局：首 4 字节 float32 = 该股票在全局日历中的起始下标，
  其后每个交易日一个 float32（停牌/缺失为 NaN）
- provider_uri 需以 {"day": <目录>} 传入 qlib.init
"""

from __future__ import annotations

from pathlib import Path

import numpy as np
import pandas as pd

# qlib 消费侧字段（Alpha158 依赖：close/open/high/low/volume/amount/vwap/factor）
DUMP_FIELDS = ["open", "high", "low", "close", "volume", "amount", "vwap", "factor"]


def _synth_vwap(daily: pd.DataFrame) -> pd.Series:
    """vwap = amount / volume（真实口径）；volume 为 0 时取 close。"""
    vol = daily["volume"].replace(0, pd.NA)
    vwap = daily["amount"] / vol
    return vwap.fillna(daily["close"])


def build_qlib_dataset(
    daily: pd.DataFrame,
    out_dir: Path,
    *,
    fields: tuple[str, ...] = DUMP_FIELDS,
    add_vwap: bool = True,
) -> Path:
    """把长表行情落盘为 qlib 数据集目录；返回 out_dir（provider_uri）。"""
    if daily.empty:
        raise ValueError("无行情数据可导出 Qlib 数据集")
    out_dir = Path(out_dir)
    cal_file = out_dir / "calendars" / "day.txt"
    cal_file.parent.mkdir(parents=True, exist_ok=True)
    (out_dir / "instruments").mkdir(exist_ok=True)
    feat_root = out_dir / "features"

    df = daily.copy()
    if add_vwap and "amount" in df.columns and "volume" in df.columns:
        df["vwap"] = _synth_vwap(df)

    calendar = sorted(df["trade_date"].unique())
    if not isinstance(calendar[0], (pd.Timestamp, np.datetime64)):
        calendar = pd.to_datetime(calendar).sort_values().tolist()
    cal_idx = {day: i for i, day in enumerate(calendar)}
    cal_file.write_text(
        "\n".join(pd.Timestamp(d).strftime("%Y-%m-%d") for d in calendar), encoding="utf-8"
    )

    inst_lines: list[str] = []
    for symbol, group in df.groupby("symbol"):
        symbol = str(symbol).lower()
        dates = sorted(pd.to_datetime(group["trade_date"]))
        start_idx = cal_idx[dates[0]]
        end_idx = cal_idx[dates[-1]]
        inst_lines.append(
            f"{symbol}\t{pd.Timestamp(dates[0]).strftime('%Y-%m-%d')}"
            f"\t{pd.Timestamp(dates[-1]).strftime('%Y-%m-%d')}"
        )
        sym_dir = feat_root / symbol
        sym_dir.mkdir(parents=True, exist_ok=True)
        per_date = group.set_index(pd.to_datetime(group["trade_date"]))
        for field in fields:
            if field not in per_date.columns:
                continue
            series = per_date[field]
            values = np.full(len(calendar), np.nan, dtype=np.float32)
            for day, val in series.items():
                if pd.isna(val):
                    continue
                try:
                    values[cal_idx[day]] = float(val)
                except (TypeError, ValueError, KeyError):
                    continue
            payload = np.hstack(
                [np.asarray([float(start_idx)], dtype=np.float32), values[start_idx : end_idx + 1]]
            ).astype("<f")
            (sym_dir / f"{field}.day.bin").write_bytes(payload.tobytes())

    (out_dir / "instruments" / "all.txt").write_text("\n".join(inst_lines), encoding="utf-8")
    return out_dir