"""Tiny HF Market Data client shared by the hfmd-* skills (requests + pandas only).

Copied verbatim into every skill's `scripts/` folder by `skills/scripts/build_skills.py`
so each skill stays self-contained. Do not edit the copies — edit `skills/_shared/hfmd.py`.

Environment: HFMD_API_KEY (optional; free API key = 120 req/min, keyless mode has low hourly limits = 30 req/h;
             higher limits on request by e-mail to contact@spboucher.ai, also free),
             HFMD_BASE_URL (default https://www.hfmarketdata.io).
"""
from __future__ import annotations

import os
import sys
import time
from typing import Any

import pandas as pd
import requests

BASE_URL = os.environ.get("HFMD_BASE_URL", "https://www.hfmarketdata.io").rstrip("/")
API_KEY = os.environ.get("HFMD_API_KEY", "").strip()
UA = "hfmd-skills/1.0 (+https://www.hfmarketdata.io/integrations/skills)"

TIMEFRAMES = {"1m": "1min", "5m": "5min", "30m": "30min", "1h": "1hour", "1d": "1day",
              "1min": "1min", "5min": "5min", "30min": "30min", "1hour": "1hour", "1day": "1day"}
V2_INTERVAL = {"1min": "1m", "5min": "5m", "30min": "30m", "1hour": "1h", "1day": "1d"}


class HfmdError(RuntimeError):
    def __init__(self, status: int, code: str, message: str, url: str):
        super().__init__(f"HF Market Data error {status} {code}: {message} ({url})")
        self.status, self.code, self.url = status, code, url


_session = requests.Session()
_session.headers.update({"Accept": "application/json", "User-Agent": UA})
if API_KEY:
    _session.headers["Authorization"] = f"Bearer {API_KEY}"


def log(msg: str) -> None:
    print(msg, file=sys.stderr)


def get(path: str, params: dict[str, Any] | None = None, *, retries: int = 3) -> tuple[Any, dict[str, str]]:
    """GET a JSON endpoint. Returns (body, headers). Retries on 429 honouring Retry-After."""
    url = f"{BASE_URL}{path}"
    p = {k: (",".join(v) if isinstance(v, (list, tuple)) else v) for k, v in (params or {}).items() if v not in (None, "")}
    p.setdefault("format", "json")
    for attempt in range(retries + 1):
        r = _session.get(url, params=p, timeout=120)
        if r.status_code == 429 and attempt < retries:
            wait = float(r.headers.get("Retry-After", "5"))
            log(f"429 rate limited — sleeping {wait:.0f}s (set HFMD_API_KEY for 120 req/min)")
            time.sleep(min(wait, 120))
            continue
        if r.status_code >= 400:
            try:
                body = r.json()
            except ValueError:
                body = {}
            err = body.get("error") or {}
            raise HfmdError(r.status_code, err.get("code", "HTTP_ERROR"), err.get("message") or body.get("detail") or r.text[:200], r.url)
        rem = r.headers.get("X-RateLimit-Remaining-Requests")
        if rem is not None:
            log(f"rate limit: {rem}/{r.headers.get('X-RateLimit-Limit-Requests')} requests left")
        return r.json(), dict(r.headers)
    raise HfmdError(429, "RATE_LIMIT_EXCEEDED", "gave up after retries", url)


def rows_of(body: Any) -> list[dict]:
    """Rows from either envelope: v1 {count,data} or v2 {data,meta}."""
    if isinstance(body, list):
        return body
    if isinstance(body, dict):
        data = body.get("data")
        if isinstance(data, list):
            return data
    return []


def meta_of(body: Any) -> dict:
    if isinstance(body, dict):
        m = dict(body.get("meta") or {})
        if "count" in body and "count" not in m:
            m["count"] = body["count"]
        return m
    return {}


def to_df(body: Any, time_col: str | None = "datetime") -> pd.DataFrame:
    df = pd.DataFrame(rows_of(body))
    if time_col and time_col in df.columns:
        df[time_col] = pd.to_datetime(df[time_col])
        df = df.sort_values(time_col).reset_index(drop=True)
    return df


def bars(asset: str, ticker: str, timeframe: str = "1day", start: str | None = None, end: str | None = None,
         adjustment: str | None = None, limit: int = 50_000) -> pd.DataFrame:
    """v1 bars (stock/etf/crypto/index/fx, or futures = vendor continuous). Paginates by date when needed."""
    tf = TIMEFRAMES.get(timeframe, timeframe)
    frames: list[pd.DataFrame] = []
    cursor_start = start
    while True:
        body, _ = get(f"/v1/bars/{asset}/{ticker}", {"timeframe": tf, "start": cursor_start, "end": end, "adjustment": adjustment, "limit": limit, "order": "asc"})
        df = to_df(body)
        if df.empty:
            break
        frames.append(df)
        if len(df) < limit:
            break
        last = df["datetime"].iloc[-1]
        cursor_start = (last + pd.Timedelta(minutes=1)).strftime("%Y-%m-%dT%H:%M") if tf != "1day" else (last + pd.Timedelta(days=1)).strftime("%Y-%m-%d")
    out = pd.concat(frames, ignore_index=True) if frames else pd.DataFrame(columns=["ticker", "datetime", "open", "high", "low", "close", "volume"])
    return out.drop_duplicates("datetime").reset_index(drop=True)


def contract_bars(symbol: str, timeframe: str = "1day", start: str | None = None, end: str | None = None, limit: int = 50_000) -> pd.DataFrame:
    """v2 individual futures contract bars (ESZ25…)."""
    tf = TIMEFRAMES.get(timeframe, timeframe)
    body, _ = get(f"/v1/futures/contract/{symbol}/bars", {"interval": V2_INTERVAL.get(tf, "1d"), "from": start, "to": end, "limit": limit})
    return to_df(body)


def continuous(root: str, roll: str = "volume", adjust: str = "back_adjusted", depth: int = 1, timeframe: str = "1day",
               start: str | None = None, end: str | None = None, limit: int = 50_000) -> tuple[pd.DataFrame, dict]:
    """v2 server-built continuous series. Returns (df, meta) — meta['roll_dates'] lists the rolls."""
    tf = TIMEFRAMES.get(timeframe, timeframe)
    body, _ = get(f"/v1/futures/{root}/continuous", {"roll": roll, "adjust": adjust, "depth": depth, "interval": V2_INTERVAL.get(tf, "1d"), "from": start, "to": end, "limit": limit})
    return to_df(body), meta_of(body)


def contracts(root: str, status: str | None = None) -> pd.DataFrame:
    body, _ = get(f"/v1/futures/{root}/contracts", {"status": status, "limit": 1000})
    return to_df(body, time_col=None)


def term_structure(root: str, as_of: str | None = None) -> tuple[pd.DataFrame, dict]:
    body, _ = get(f"/v1/futures/{root}/term-structure", {"as_of": as_of})
    return to_df(body, time_col=None), meta_of(body)


def screener(filters: str, sort: str | None = None, as_of: str | None = None, limit: int = 100) -> tuple[pd.DataFrame, dict]:
    body, _ = get("/v1/fundamentals/screener", {"filters": filters, "sort": sort, "as_of": as_of, "limit": limit})
    return to_df(body, time_col=None), meta_of(body)


def ratios(ticker: str) -> dict:
    body, _ = get(f"/v1/fundamentals/{ticker}/ratios")
    d = body.get("data", body) if isinstance(body, dict) else body
    return d if isinstance(d, dict) else (d[0] if d else {})


def statements(ticker: str, statement: str = "income", period: str = "quarterly", limit: int = 12) -> pd.DataFrame:
    body, _ = get(f"/v1/fundamentals/{ticker}/statements", {"statement": statement, "period": period, "limit": limit})
    return to_df(body, time_col=None)