point_in_time_feature_collector.py

#!/usr/bin/env python3
"""Collect read-only Bitget RWA features for future leakage-safe research.

Historical responses are useful for coverage diagnostics, but bars observed
before this collector was first started remain ineligible for strategy
selection because they overlap the already locked final holdout.
"""

from __future__ import annotations

import argparse
import json
import math
import subprocess
import time
import urllib.parse
from datetime import datetime, timezone
from pathlib import Path
from typing import Any

import pandas as pd

from common import BASE_DIR, BITGET_API_BASE, BITGET_PRODUCT_TYPE, RESULTS_DIR
from integrity_io import atomic_write_json


WATCHLIST_PATH = BASE_DIR / "data" / "bitget_us_equity_contract_watchlist.csv"
FEATURE_DIR = BASE_DIR / "data" / "point_in_time_features" / "bitget_rwa"
STATE_PATH = FEATURE_DIR / "collector_state.json"
STATUS_PATH = RESULTS_DIR / "point_in_time_feature_status.json"
SUMMARY_PATH = RESULTS_DIR / "point_in_time_feature_summary.csv"
REPORT_PATH = RESULTS_DIR / "point_in_time_feature_report.md"

MIN_READY_TICKERS = 5
MIN_FUNDING_POINTS = 60
MIN_NONZERO_FUNDING_POINTS = 5
MIN_OI_POINTS = 160
MIN_OI_SPAN_DAYS = 30.0
MIN_PROSPECTIVE_HOURLY_BARS = 240
MIN_PROSPECTIVE_SPAN_DAYS = 30.0


def now_iso() -> str:
    return datetime.now(timezone.utc).isoformat(timespec="seconds")


def safe_float(value: Any, default: float = 0.0) -> float:
    try:
        number = float(value)
        return number if math.isfinite(number) else float(default)
    except Exception:
        return float(default)


def read_json(path: Path, default: Any) -> Any:
    try:
        return json.loads(path.read_text(encoding="utf-8-sig"))
    except Exception:
        return default


def fetch_json(path: str, params: dict[str, Any], timeout: int = 20, retries: int = 3) -> dict[str, Any]:
    query = urllib.parse.urlencode(params)
    completed = subprocess.run(
        [
            "curl",
            "-fsS",
            "--retry",
            str(max(0, int(retries) - 1)),
            "--retry-all-errors",
            "--connect-timeout",
            "5",
            "--max-time",
            str(max(6, int(timeout))),
            "-H",
            "User-Agent: Mozilla/5.0 quant-machine-point-in-time/1.0",
            f"{BITGET_API_BASE}{path}?{query}",
        ],
        capture_output=True,
        text=True,
        timeout=max(10, int(timeout) * max(1, int(retries))),
        check=False,
    )
    if completed.returncode != 0:
        raise RuntimeError(f"Bitget curl failed {path}: {completed.stderr.strip()}")
    payload = json.loads(completed.stdout)
    if str(payload.get("code")) != "00000":
        raise RuntimeError(f"Bitget {payload.get('code')}: {payload.get('msg')}")
    return payload


def read_frame(path: Path) -> pd.DataFrame:
    try:
        return pd.read_csv(path)
    except Exception:
        return pd.DataFrame()


def merge_frame(path: Path, rows: list[dict[str, Any]], timestamp_column: str) -> pd.DataFrame:
    current = read_frame(path)
    incoming = pd.DataFrame(rows)
    if current.empty:
        merged = incoming
    elif incoming.empty:
        merged = current
    else:
        merged = pd.concat([current, incoming], ignore_index=True)
    if merged.empty:
        return merged
    merged[timestamp_column] = pd.to_datetime(
        merged[timestamp_column], utc=True, errors="coerce", format="mixed"
    )
    sort_columns = [timestamp_column]
    if "observed_at" in merged.columns:
        merged["observed_at"] = pd.to_datetime(
            merged["observed_at"], utc=True, errors="coerce", format="mixed"
        )
        sort_columns.append("observed_at")
    merged = merged.dropna(subset=[timestamp_column]).sort_values(sort_columns, kind="mergesort")
    merged = merged.drop_duplicates(subset=[timestamp_column], keep="last").reset_index(drop=True)
    path.parent.mkdir(parents=True, exist_ok=True)
    merged.to_csv(path, index=False, encoding="utf-8-sig")
    return merged


def refresh_due(path: Path, maximum_age_seconds: float = 6 * 3600) -> bool:
    if not path.exists():
        return True
    return time.time() - path.stat().st_mtime >= float(maximum_age_seconds)


def load_watchlist(max_tickers: int) -> list[dict[str, Any]]:
    frame = read_frame(WATCHLIST_PATH)
    if frame.empty:
        raise RuntimeError("Bitget美股合约观察清单不存在")
    frame["usdt_volume_24h"] = pd.to_numeric(frame.get("usdt_volume_24h"), errors="coerce").fillna(0.0)
    frame = frame.sort_values(["usdt_volume_24h", "ticker"], ascending=[False, True])
    return frame.head(max(1, int(max_tickers))).to_dict(orient="records")


def funding_rows(symbol: str, observed_at: str) -> list[dict[str, Any]]:
    payload = fetch_json(
        "/api/v2/mix/market/history-fund-rate",
        {"symbol": symbol, "productType": BITGET_PRODUCT_TYPE, "pageSize": 100, "pageNo": 1},
    )
    return [
        {
            "datetime": pd.to_datetime(int(row.get("fundingTime")), unit="ms", utc=True).isoformat(),
            "funding_rate": safe_float(row.get("fundingRate")),
            "observed_at": observed_at,
            "source": "Bitget public history-fund-rate",
        }
        for row in payload.get("data") or []
        if row.get("fundingTime")
    ]


def oi_rows(symbol: str, observed_at: str) -> list[dict[str, Any]]:
    payload = fetch_json(
        "/api/v2/mix/market/open-interest",
        {"symbol": symbol, "productType": BITGET_PRODUCT_TYPE},
    )
    data = payload.get("data") or {}
    timestamp = data.get("ts")
    rows = data.get("openInterestList") or []
    if not timestamp or not rows:
        return []
    observed_hour = pd.to_datetime(int(timestamp), unit="ms", utc=True).floor("h")
    return [
        {
            "datetime": observed_hour.isoformat(),
            "open_interest": safe_float(row.get("size")),
            "observed_at": observed_at,
            "source": "Bitget public open-interest snapshot",
        }
        for row in rows
        if str(row.get("symbol", "")).upper() == symbol.upper()
    ]


def hourly_rows(symbol: str, observed_at: str) -> list[dict[str, Any]]:
    payload = fetch_json(
        "/api/v2/mix/market/candles",
        {"symbol": symbol, "productType": BITGET_PRODUCT_TYPE, "granularity": "1H", "limit": 1000},
    )
    rows = []
    observed = pd.Timestamp(observed_at)
    observed = observed.tz_localize("UTC") if observed.tzinfo is None else observed.tz_convert("UTC")
    for row in payload.get("data") or []:
        if not isinstance(row, list) or len(row) < 7:
            continue
        bar_open = pd.to_datetime(int(row[0]), unit="ms", utc=True)
        if bar_open + pd.Timedelta(hours=1) > observed:
            continue
        rows.append(
            {
                "datetime": bar_open.isoformat(),
                "open": safe_float(row[1]),
                "high": safe_float(row[2]),
                "low": safe_float(row[3]),
                "close": safe_float(row[4]),
                "volume": safe_float(row[5]),
                "volume_usd": safe_float(row[6]),
                "observed_at": observed_at,
                "source": "Bitget public 1H candles",
            }
        )
    return rows


def span_days(frame: pd.DataFrame, column: str = "datetime") -> float:
    if frame.empty or column not in frame:
        return 0.0
    values = pd.to_datetime(frame[column], utc=True, errors="coerce").dropna()
    return max(0.0, (values.max() - values.min()).total_seconds() / 86400.0) if len(values) > 1 else 0.0


def summarize_ticker(ticker: str, collector_started_at: str) -> dict[str, Any]:
    prefix = FEATURE_DIR / ticker
    funding = read_frame(prefix.with_name(f"{ticker}_funding.csv"))
    oi = read_frame(prefix.with_name(f"{ticker}_oi.csv"))
    hourly = read_frame(prefix.with_name(f"{ticker}_1h.csv"))
    funding_values = pd.to_numeric(funding.get("funding_rate", pd.Series(dtype=float)), errors="coerce").dropna()
    funding_times = pd.to_datetime(funding.get("datetime", pd.Series(dtype=str)), utc=True, errors="coerce")
    hourly_times = pd.to_datetime(hourly.get("datetime", pd.Series(dtype=str)), utc=True, errors="coerce")
    start = pd.Timestamp(collector_started_at)
    prospective_funding = funding.loc[funding_times > start].copy() if not funding.empty else pd.DataFrame()
    prospective_funding_values = pd.to_numeric(prospective_funding.get("funding_rate", pd.Series(dtype=float)), errors="coerce").dropna()
    prospective = hourly.loc[hourly_times > start].copy() if not hourly.empty else pd.DataFrame()
    return {
        "ticker": ticker,
        "funding_points": int(len(funding_values)),
        "funding_nonzero_points": int((funding_values.abs() > 1e-10).sum()),
        "funding_std": safe_float(funding_values.std(ddof=0)) if len(funding_values) else 0.0,
        "prospective_funding_points": int(len(prospective_funding_values)),
        "prospective_funding_nonzero_points": int((prospective_funding_values.abs() > 1e-10).sum()),
        "prospective_funding_std": safe_float(prospective_funding_values.std(ddof=0)) if len(prospective_funding_values) else 0.0,
        "oi_points": int(len(oi)),
        "oi_span_days": round(span_days(oi), 2),
        "hourly_bars": int(len(hourly)),
        "hourly_span_days": round(span_days(hourly), 2),
        "prospective_hourly_bars": int(len(prospective)),
        "prospective_hourly_span_days": round(span_days(prospective), 2),
    }


def assess_strategy_data(strategy: str, ticker_rows: list[dict[str, Any]]) -> dict[str, Any]:
    if strategy == "4":
        covered = sum(int(row.get("funding_points", 0)) >= MIN_FUNDING_POINTS for row in ticker_rows)
        usable = sum(
            int(row.get("prospective_funding_points", 0)) >= MIN_FUNDING_POINTS
            and int(row.get("prospective_funding_nonzero_points", 0)) >= MIN_NONZERO_FUNDING_POINTS
            and safe_float(row.get("prospective_funding_std")) > 1e-8
            for row in ticker_rows
        )
        historical_nonzero = sum(int(row.get("funding_nonzero_points", 0)) >= MIN_NONZERO_FUNDING_POINTS for row in ticker_rows)
        if usable >= MIN_READY_TICKERS:
            status = "ready_for_training_only"
        elif covered >= MIN_READY_TICKERS and historical_nonzero == 0:
            status = "blocked_non_informative_funding"
        else:
            status = "collecting_prospective_funding"
        reason = "资金费率有时间戳但几乎全为0，无法形成可识别因子" if status == "blocked_non_informative_funding" else "已有历史覆盖，但锁定样本段不参与选择；只累积采集启动后的新资金费率"
    elif strategy == "6":
        covered = len(ticker_rows)
        usable = sum(
            int(row.get("oi_points", 0)) >= MIN_OI_POINTS
            and safe_float(row.get("oi_span_days")) >= MIN_OI_SPAN_DAYS
            for row in ticker_rows
        )
        status = "ready_for_training_only" if usable >= MIN_READY_TICKERS else "collecting_point_in_time_oi"
        reason = "OI公开接口只提供当前快照，需本地持续积累至30日"
    elif strategy == "8":
        covered = sum(int(row.get("hourly_bars", 0)) > 0 for row in ticker_rows)
        usable = sum(
            int(row.get("prospective_hourly_bars", 0)) >= MIN_PROSPECTIVE_HOURLY_BARS
            and safe_float(row.get("prospective_hourly_span_days")) >= MIN_PROSPECTIVE_SPAN_DAYS
            for row in ticker_rows
        )
        status = "ready_for_training_only" if usable >= MIN_READY_TICKERS else "collecting_prospective_intraday"
        reason = "已下载1H覆盖数据，但锁定样本段不参与选择；只累积采集启动后的新数据"
    else:
        raise ValueError(f"Unsupported strategy: {strategy}")
    return {
        "strategy": strategy,
        "status": status,
        "covered_tickers": int(covered),
        "usable_tickers": int(usable),
        "minimum_ready_tickers": MIN_READY_TICKERS,
        "training_data_ready": status == "ready_for_training_only",
        "promotion_allowed": False,
        "reason": reason,
    }


def write_status(started_at: str, ticker_rows: list[dict[str, Any]], errors: list[dict[str, str]]) -> dict[str, Any]:
    robots = {strategy: assess_strategy_data(strategy, ticker_rows) for strategy in ("4", "6", "8")}
    payload = {
        "generated_at": now_iso(),
        "collector_started_at": started_at,
        "paper_trading_only": True,
        "locked_holdout_consumed_for_selection": False,
        "ticker_count": len(ticker_rows),
        "robots": robots,
        "tickers": ticker_rows,
        "errors": errors,
    }
    STATUS_PATH.parent.mkdir(parents=True, exist_ok=True)
    atomic_write_json(STATUS_PATH, payload)
    pd.DataFrame(ticker_rows).to_csv(SUMMARY_PATH, index=False, encoding="utf-8-sig")
    report = [
        "# 点时因子数据成熟度",
        "",
        f"- 采集启动：{started_at}",
        "- 历史覆盖只作数据诊断；采集启动前落入锁定样本段的数据不用于参数或因子选择。",
        "- 本采集器只读 Bitget 公开市场数据，不下单。",
        "",
        "| 机器人 | 状态 | 覆盖标的 | 可用标的 | 说明 |",
        "|---|---|---:|---:|---|",
    ]
    for strategy in ("4", "6", "8"):
        item = robots[strategy]
        report.append(f"| {strategy} | {item['status']} | {item['covered_tickers']} | {item['usable_tickers']} | {item['reason']} |")
    report.extend(["", "不构成投资建议；只用于虚拟盘与研究。", ""])
    REPORT_PATH.write_text("\n".join(report), encoding="utf-8")
    return payload


def run_once(max_tickers: int = 40) -> dict[str, Any]:
    FEATURE_DIR.mkdir(parents=True, exist_ok=True)
    state = read_json(STATE_PATH, {})
    started_at = state.get("collector_started_at") or now_iso()
    observed_at = now_iso()
    errors: list[dict[str, str]] = []
    watchlist = load_watchlist(max_tickers)
    for item in watchlist:
        ticker = str(item.get("ticker", "")).upper().strip()
        symbol = str(item.get("symbol") or f"{ticker}USDT").upper()
        if not ticker:
            continue
        for feature, loader, suffix, timestamp in (
            ("funding", funding_rows, "funding", "datetime"),
            ("oi", oi_rows, "oi", "datetime"),
            ("hourly", hourly_rows, "1h", "datetime"),
        ):
            try:
                target = FEATURE_DIR / f"{ticker}_{suffix}.csv"
                if feature in {"funding", "hourly"} and not refresh_due(target):
                    continue
                rows = loader(symbol, observed_at)
                merge_frame(target, rows, timestamp)
            except Exception as exc:
                errors.append({"ticker": ticker, "feature": feature, "error": str(exc)})
            time.sleep(0.06)
    ticker_rows = [summarize_ticker(str(item.get("ticker", "")).upper(), started_at) for item in watchlist]
    state.update({"collector_started_at": started_at, "last_completed_at": now_iso(), "paper_trading_only": True})
    atomic_write_json(STATE_PATH, state)
    return write_status(started_at, ticker_rows, errors)


def main() -> None:
    parser = argparse.ArgumentParser()
    parser.add_argument("--loop", action="store_true")
    parser.add_argument("--interval-minutes", type=float, default=15.0)
    parser.add_argument("--max-tickers", type=int, default=40)
    args = parser.parse_args()
    while True:
        payload = run_once(max_tickers=args.max_tickers)
        print(json.dumps({"generated_at": payload["generated_at"], "ticker_count": payload["ticker_count"], "robots": payload["robots"], "errors": len(payload["errors"])}, ensure_ascii=False), flush=True)
        if not args.loop:
            break
        time.sleep(max(60.0, float(args.interval_minutes) * 60.0))


if __name__ == "__main__":
    main()