"""中控振幅统计:OKX 指数(可降级永续)按时段切窗,点数口径.

仅只读行情;不触及下单链路.
"""
from __future__ import annotations

import csv
import io
import statistics
from datetime import date, datetime, timedelta
from typing import Any, Callable, Optional
from zoneinfo import ZoneInfo

import httpx

APP_TZ = ZoneInfo("Asia/Shanghai")
END_HOUR = 16
EXCHANGE = "okx"
TIMEFRAME = "1H"

SYMBOLS: dict[str, dict[str, str]] = {
    "eth": {
        "label": "ETH",
        "index_inst": "ETH-USD",
        "swap_inst": "ETH-USDT-SWAP",
    },
    "btc": {
        "label": "BTC",
        "index_inst": "BTC-USD",
        "swap_inst": "BTC-USDT-SWAP",
    },
}

PERIOD_DAYS: dict[str, int] = {
    "1m": 30,
    "2m": 60,
    "3m": 90,
    "6m": 180,
    "1y": 365,
}

OKX_INDEX_CANDLES = "https://www.okx.com/api/v5/market/index-candles"
OKX_HISTORY_INDEX_CANDLES = "https://www.okx.com/api/v5/market/history-index-candles"
OKX_SWAP_CANDLES = "https://www.okx.com/api/v5/market/candles"
OKX_HISTORY_SWAP_CANDLES = "https://www.okx.com/api/v5/market/history-candles"


def normalize_symbol(raw: str) -> str:
    s = (raw or "").strip().lower()
    if s in ("eth", "ethereum"):
        return "eth"
    if s in ("btc", "bitcoin"):
        return "btc"
    raise ValueError("symbol 仅支持 eth / btc")


def resolve_sample_days(period: str, custom_days: Any = None) -> int:
    p = (period or "2m").strip().lower()
    if p == "custom":
        try:
            n = int(custom_days)
        except (TypeError, ValueError):
            raise ValueError("自定义天数无效") from None
        return max(7, min(400, n))
    if p not in PERIOD_DAYS:
        raise ValueError("周期无效")
    return PERIOD_DAYS[p]


def window_bounds_for_settlement(settlement: date, start_hour: int) -> tuple[datetime, datetime]:
    """返回 [start, end) 的本地时刻;end 为结算日 16:00."""
    if not (0 <= int(start_hour) <= 23):
        raise ValueError("起点须为 0-23 整点")
    end = datetime(settlement.year, settlement.month, settlement.day, END_HOUR, 0, 0, tzinfo=APP_TZ)
    sh = int(start_hour)
    if sh >= END_HOUR:
        prev = settlement - timedelta(days=1)
        start = datetime(prev.year, prev.month, prev.day, sh, 0, 0, tzinfo=APP_TZ)
    else:
        start = datetime(settlement.year, settlement.month, settlement.day, sh, 0, 0, tzinfo=APP_TZ)
    return start, end


def list_settlement_dates(*, sample_days: int, now: Optional[datetime] = None) -> list[date]:
    """最近 sample_days 个已收窗结算日(不含进行中的今天未到 16:00)."""
    now = now or datetime.now(APP_TZ)
    if now.tzinfo is None:
        now = now.replace(tzinfo=APP_TZ)
    else:
        now = now.astimezone(APP_TZ)
    today = now.date()
    today_end = datetime(today.year, today.month, today.day, END_HOUR, 0, 0, tzinfo=APP_TZ)
    latest = today if now >= today_end else today - timedelta(days=1)
    return [latest - timedelta(days=i) for i in range(int(sample_days))]


def _safe_float(v: Any) -> Optional[float]:
    try:
        if v is None or v == "":
            return None
        return float(v)
    except (TypeError, ValueError):
        return None


def bars_to_map(bars: list[dict[str, Any]]) -> dict[int, dict[str, float]]:
    """open_time_ms -> {o,h,l,c}."""
    m: dict[int, dict[str, float]] = {}
    for b in bars or []:
        if not isinstance(b, dict):
            continue
        ts = b.get("ts")
        if ts is None:
            ts = b.get("open_time_ms")
        try:
            ts_i = int(ts)
        except (TypeError, ValueError):
            continue
        o = _safe_float(b.get("o") if "o" in b else b.get("open"))
        h = _safe_float(b.get("h") if "h" in b else b.get("high"))
        l = _safe_float(b.get("l") if "l" in b else b.get("low"))
        c = _safe_float(b.get("c") if "c" in b else b.get("close"))
        if None in (o, h, l, c):
            continue
        m[ts_i] = {"o": float(o), "h": float(h), "l": float(l), "c": float(c)}
    return m


def compute_day_row(
    settlement: date,
    start_hour: int,
    bar_map: dict[int, dict[str, float]],
) -> Optional[dict[str, Any]]:
    start, end = window_bounds_for_settlement(settlement, start_hour)
    start_ms = int(start.timestamp() * 1000)
    # 1H 棒覆盖 [T, T+1h);窗终点 16:00 用 15:00 棒的 close
    last_bar_ms = int((end - timedelta(hours=1)).timestamp() * 1000)
    if start_ms not in bar_map or last_bar_ms not in bar_map:
        return None
    opens = bar_map[start_ms]["o"]
    close = bar_map[last_bar_ms]["c"]
    hi = bar_map[start_ms]["h"]
    lo = bar_map[start_ms]["l"]
    t = start_ms
    while t <= last_bar_ms:
        b = bar_map.get(t)
        if b:
            hi = max(hi, b["h"])
            lo = min(lo, b["l"])
        t += 3600 * 1000
    up = hi - opens
    down = opens - lo
    amp = hi - lo
    change = close - opens
    wd = settlement.weekday()  # Mon=0 … Sun=6
    is_we = wd >= 5
    return {
        "settlement_day": settlement.isoformat(),
        "window_start": start.strftime("%Y-%m-%d %H:%M"),
        "window_end": end.strftime("%Y-%m-%d %H:%M"),
        "weekday": wd,
        "weekday_label": "六" if wd == 5 else ("日" if wd == 6 else ""),
        "is_weekend": is_we,
        "open": round(opens, 4),
        "high": round(hi, 4),
        "low": round(lo, 4),
        "close": round(close, 4),
        "up_points": round(up, 4),
        "down_points": round(down, 4),
        "amplitude": round(amp, 4),
        "change": round(change, 4),
    }


def normalize_straddle_premium(raw: Any) -> Optional[float]:
    """双边权利金(点数).空/≤0 表示不做跨式对照."""
    if raw is None or raw == "":
        return None
    try:
        v = float(raw)
    except (TypeError, ValueError):
        raise ValueError("双边权利金须为数字") from None
    if v <= 0:
        return None
    return v


def normalize_take_profit(raw: Any) -> Optional[float]:
    """止盈点.空/≤0 表示不止盈,有效波动用 |涨跌|."""
    if raw is None or raw == "":
        return None
    try:
        v = float(raw)
    except (TypeError, ValueError):
        raise ValueError("止盈点须为数字") from None
    if v <= 0:
        return None
    return v


def normalize_weekend_filter(raw: Any) -> str:
    """all | exclude | only;默认全部."""
    s = (str(raw) if raw is not None else "all").strip().lower()
    if s in ("", "all", "全部"):
        return "all"
    if s in ("exclude", "exclude_weekend", "no_weekend", "排除周末"):
        return "exclude"
    if s in ("only", "weekend_only", "only_weekend", "仅周末"):
        return "only"
    raise ValueError("周末筛选须为 all / exclude / only")


def filter_weekend_rows(rows: list[dict[str, Any]], weekend_filter: Any = "all") -> list[dict[str, Any]]:
    mode = normalize_weekend_filter(weekend_filter)
    if mode == "all":
        return list(rows or [])
    out: list[dict[str, Any]] = []
    for r in rows or []:
        is_we = bool(r.get("is_weekend"))
        if "is_weekend" not in r and r.get("settlement_day"):
            try:
                is_we = date.fromisoformat(str(r["settlement_day"])).weekday() >= 5
            except ValueError:
                is_we = False
        if mode == "exclude" and is_we:
            continue
        if mode == "only" and not is_we:
            continue
        out.append(r)
    return out


def effective_move_points(row: dict[str, Any], take_profit: Optional[float]) -> float:
    """触达止盈(≥)用止盈点,否则用 |涨跌|."""
    abs_chg = abs(float(row.get("change") or 0))
    if take_profit is None:
        return abs_chg
    tp = float(take_profit)
    up = float(row.get("up_points") or 0)
    down = float(row.get("down_points") or 0)
    if up >= tp or down >= tp:
        return tp
    return abs_chg


def enrich_rows_pnl(
    rows: list[dict[str, Any]],
    *,
    straddle_premium: Optional[float] = None,
    take_profit: Optional[float] = None,
) -> list[dict[str, Any]]:
    """为日表附加有效波动 / 是否触达止盈 / 收益(有权利金时)."""
    prem = normalize_straddle_premium(straddle_premium)
    tp = normalize_take_profit(take_profit)
    out: list[dict[str, Any]] = []
    for r in rows or []:
        item = dict(r)
        if "is_weekend" not in item and item.get("settlement_day"):
            try:
                wd = date.fromisoformat(str(item["settlement_day"])).weekday()
                item["weekday"] = wd
                item["weekday_label"] = "六" if wd == 5 else ("日" if wd == 6 else "")
                item["is_weekend"] = wd >= 5
            except ValueError:
                item.setdefault("weekday_label", "")
                item.setdefault("is_weekend", False)
        move = effective_move_points(item, tp)
        hit = False
        if tp is not None:
            hit = float(item.get("up_points") or 0) >= tp or float(item.get("down_points") or 0) >= tp
        item["effective_move"] = round(move, 4)
        item["take_profit_hit"] = hit
        item["profit"] = round(move - prem, 4) if prem is not None else None
        out.append(item)
    return out


def straddle_long_stats(
    rows: list[dict[str, Any]],
    premium: float,
    *,
    take_profit: Any = None,
) -> dict[str, Any]:
    """买跨:越过权利金用严格 >;收益=有效波动−权利金(止盈≥触达用止盈点,否则|涨跌|)."""
    prem = float(premium)
    if prem <= 0:
        raise ValueError("双边权利金须 > 0")
    tp = normalize_take_profit(take_profit)
    enriched = enrich_rows_pnl(rows, straddle_premium=prem, take_profit=tp)
    if not enriched:
        return {
            "side": "long_straddle",
            "premium": prem,
            "take_profit": tp,
            "sample_count": 0,
            "up_exceed_days": 0,
            "up_exceed_ratio": None,
            "down_exceed_days": 0,
            "down_exceed_ratio": None,
            "abs_change_exceed_days": 0,
            "abs_change_exceed_ratio": None,
            "tp_hit_days": 0,
            "tp_hit_ratio": None,
            "pnl_total": None,
            "pnl_avg": None,
            "win_days": 0,
            "win_ratio": None,
            "pnl_max": None,
            "pnl_min": None,
        }
    n = len(enriched)
    up_ex = sum(1 for r in enriched if float(r["up_points"]) > prem)
    down_ex = sum(1 for r in enriched if float(r["down_points"]) > prem)
    abs_ex = sum(1 for r in enriched if abs(float(r["change"])) > prem)
    tp_hits = sum(1 for r in enriched if r.get("take_profit_hit"))
    pnls = [float(r["profit"]) for r in enriched if r.get("profit") is not None]
    win = sum(1 for p in pnls if p > 0)
    return {
        "side": "long_straddle",
        "premium": round(prem, 4),
        "take_profit": round(tp, 4) if tp is not None else None,
        "sample_count": n,
        "up_exceed_days": up_ex,
        "up_exceed_ratio": round(up_ex / n, 4),
        "down_exceed_days": down_ex,
        "down_exceed_ratio": round(down_ex / n, 4),
        "abs_change_exceed_days": abs_ex,
        "abs_change_exceed_ratio": round(abs_ex / n, 4),
        "tp_hit_days": tp_hits,
        "tp_hit_ratio": round(tp_hits / n, 4) if tp is not None else None,
        "pnl_total": round(sum(pnls), 4),
        "pnl_avg": round(statistics.fmean(pnls), 4),
        "win_days": win,
        "win_ratio": round(win / n, 4),
        "pnl_max": round(max(pnls), 4),
        "pnl_min": round(min(pnls), 4),
    }


def summarize_rows(
    rows: list[dict[str, Any]],
    *,
    straddle_premium: Any = None,
    take_profit: Any = None,
) -> dict[str, Any]:
    if not rows:
        out = {
            "sample_count": 0,
            "max_amplitude": None,
            "max_amplitude_day": None,
            "avg_amplitude": None,
            "median_amplitude": None,
            "max_up_points": None,
            "avg_up_points": None,
            "max_down_points": None,
            "avg_down_points": None,
            "up_day_ratio": None,
            "down_day_ratio": None,
            "straddle": None,
        }
        prem = normalize_straddle_premium(straddle_premium)
        if prem is not None:
            out["straddle"] = straddle_long_stats([], prem, take_profit=take_profit)
        return out
    amps = [float(r["amplitude"]) for r in rows]
    ups = [float(r["up_points"]) for r in rows]
    downs = [float(r["down_points"]) for r in rows]
    max_amp = max(amps)
    max_amp_day = next(r["settlement_day"] for r in rows if float(r["amplitude"]) == max_amp)
    up_days = sum(1 for r in rows if float(r["change"]) > 0)
    down_days = sum(1 for r in rows if float(r["change"]) < 0)
    n = len(rows)
    out: dict[str, Any] = {
        "sample_count": n,
        "max_amplitude": round(max_amp, 4),
        "max_amplitude_day": max_amp_day,
        "avg_amplitude": round(statistics.fmean(amps), 4),
        "median_amplitude": round(statistics.median(amps), 4),
        "max_up_points": round(max(ups), 4),
        "avg_up_points": round(statistics.fmean(ups), 4),
        "max_down_points": round(max(downs), 4),
        "avg_down_points": round(statistics.fmean(downs), 4),
        "up_day_ratio": round(up_days / n, 4),
        "down_day_ratio": round(down_days / n, 4),
        "straddle": None,
    }
    prem = normalize_straddle_premium(straddle_premium)
    if prem is not None:
        out["straddle"] = straddle_long_stats(rows, prem, take_profit=take_profit)
    return out


def _parse_okx_candle_row(row: list) -> Optional[dict[str, Any]]:
    if not row or len(row) < 5:
        return None
    try:
        ts = int(row[0])
        o, h, l, c = float(row[1]), float(row[2]), float(row[3]), float(row[4])
    except (TypeError, ValueError, IndexError):
        return None
    return {"ts": ts, "o": o, "h": h, "l": l, "c": c}


def fetch_okx_candles(
    *,
    url: str,
    inst_id: str,
    since_ms: int,
    until_ms: int,
    bar: str = "1H",
    client: Optional[httpx.Client] = None,
    timeout: float = 30.0,
    history_url: Optional[str] = None,
    max_pages: int = 200,
) -> list[dict[str, Any]]:
    """拉取 [since_ms, until_ms] 覆盖的 K 线(含边界).

    OKX 近期接口约仅 1440 根;更早需 history_* 端点续拉.
    """
    own = client is None
    client = client or httpx.Client(
        timeout=timeout,
        trust_env=False,
        headers={"User-Agent": "crypto_monitor-amp-stats/1.0"},
    )
    try:
        out: dict[int, dict[str, Any]] = {}
        after: Optional[str] = None
        active_url = url
        switched_history = False
        empty_streak = 0
        for _ in range(max(20, int(max_pages))):
            params: dict[str, str] = {"instId": inst_id, "bar": bar, "limit": "100"}
            if after:
                params["after"] = after
            r = client.get(active_url, params=params)
            r.raise_for_status()
            body = r.json()
            if str(body.get("code") or "") not in ("0", "0.0", ""):
                raise RuntimeError(body.get("msg") or f"OKX error {body.get('code')}")
            data = body.get("data") or []
            if not data:
                empty_streak += 1
                # 近期接口到头 → 切历史端点再试
                if (
                    history_url
                    and not switched_history
                    and after is not None
                ):
                    active_url = history_url
                    switched_history = True
                    empty_streak = 0
                    continue
                break
            empty_streak = 0
            oldest_ts = None
            newest_in_page = None
            for row in data:
                parsed = _parse_okx_candle_row(row)
                if not parsed:
                    continue
                ts = int(parsed["ts"])
                oldest_ts = ts if oldest_ts is None else min(oldest_ts, ts)
                newest_in_page = ts if newest_in_page is None else max(newest_in_page, ts)
                if ts < since_ms - 3600 * 1000:
                    continue
                if ts > until_ms + 3600 * 1000:
                    continue
                out[ts] = parsed
            if oldest_ts is None:
                break
            if oldest_ts <= since_ms:
                break
            # 无新进度时避免死循环
            if after is not None and str(oldest_ts) == after:
                if history_url and not switched_history:
                    active_url = history_url
                    switched_history = True
                    continue
                break
            after = str(oldest_ts)
            # 近期接口返回变少且仍未覆盖 since → 切历史
            if (
                history_url
                and not switched_history
                and len(data) < 100
                and oldest_ts > since_ms
            ):
                active_url = history_url
                switched_history = True
        return [out[k] for k in sorted(out.keys())]
    finally:
        if own:
            client.close()


def fetch_symbol_bars(
    symbol: str,
    *,
    since_ms: int,
    until_ms: int,
    fetch_fn: Optional[Callable[..., list[dict[str, Any]]]] = None,
) -> tuple[list[dict[str, Any]], str, str]:
    """返回 (bars, price_source_label, inst_id)."""
    key = normalize_symbol(symbol)
    meta = SYMBOLS[key]
    if fetch_fn:
        bars = fetch_fn(inst_id=meta["index_inst"], since_ms=since_ms, until_ms=until_ms)
        return bars, f"okx_index:{meta['index_inst']}", meta["index_inst"]

    try:
        bars = fetch_okx_candles(
            url=OKX_INDEX_CANDLES,
            history_url=OKX_HISTORY_INDEX_CANDLES,
            inst_id=meta["index_inst"],
            since_ms=since_ms,
            until_ms=until_ms,
        )
        if bars:
            return bars, f"okx_index:{meta['index_inst']}", meta["index_inst"]
    except Exception:
        bars = []

    bars = fetch_okx_candles(
        url=OKX_SWAP_CANDLES,
        history_url=OKX_HISTORY_SWAP_CANDLES,
        inst_id=meta["swap_inst"],
        since_ms=since_ms,
        until_ms=until_ms,
    )
    if not bars:
        raise RuntimeError("OKX 指数与永续 K 线均无数据")
    return bars, f"okx_swap:{meta['swap_inst']}", meta["swap_inst"]


def compute_amp_stats(
    *,
    symbol: str = "eth",
    start_hour: int = 16,
    period: str = "2m",
    custom_days: Any = None,
    straddle_premium: Any = None,
    take_profit: Any = None,
    weekend_filter: Any = "all",
    now: Optional[datetime] = None,
    fetch_fn: Optional[Callable[..., list[dict[str, Any]]]] = None,
) -> dict[str, Any]:
    key = normalize_symbol(symbol)
    sh = int(start_hour)
    if sh < 0 or sh > 23:
        raise ValueError("起点须为 0-23 整点")
    prem = normalize_straddle_premium(straddle_premium)
    tp = normalize_take_profit(take_profit)
    we_mode = normalize_weekend_filter(weekend_filter)
    sample_days = resolve_sample_days(period, custom_days)
    settlements = list_settlement_dates(sample_days=sample_days, now=now)
    if not settlements:
        raise RuntimeError("无可用结算日")
    # 最远窗起点
    oldest = settlements[-1]
    newest = settlements[0]
    start0, _ = window_bounds_for_settlement(oldest, sh)
    _, end1 = window_bounds_for_settlement(newest, sh)
    since_ms = int(start0.timestamp() * 1000)
    until_ms = int(end1.timestamp() * 1000)
    bars, price_source, inst_id = fetch_symbol_bars(
        key, since_ms=since_ms, until_ms=until_ms, fetch_fn=fetch_fn
    )
    bar_map = bars_to_map(bars)
    rows_all: list[dict[str, Any]] = []
    missing: list[str] = []
    for d in settlements:
        row = compute_day_row(d, sh, bar_map)
        if row is None:
            missing.append(d.isoformat())
            continue
        rows_all.append(row)
    return build_amp_result(
        rows_all=rows_all,
        symbol_key=key,
        start_hour=sh,
        period=period,
        sample_days=sample_days,
        straddle_premium=prem,
        take_profit=tp,
        weekend_filter=we_mode,
        price_source=price_source,
        inst_id=inst_id,
        missing=missing,
    )


def build_amp_result(
    *,
    rows_all: list[dict[str, Any]],
    symbol_key: str,
    start_hour: int,
    period: str,
    sample_days: int,
    straddle_premium: Any = None,
    take_profit: Any = None,
    weekend_filter: Any = "all",
    price_source: str = "",
    inst_id: str = "",
    missing: Optional[list[str]] = None,
) -> dict[str, Any]:
    prem = normalize_straddle_premium(straddle_premium)
    tp = normalize_take_profit(take_profit)
    we_mode = normalize_weekend_filter(weekend_filter)
    filtered = filter_weekend_rows(rows_all, we_mode)
    rows = enrich_rows_pnl(filtered, straddle_premium=prem, take_profit=tp)
    summary = summarize_rows(rows, straddle_premium=prem, take_profit=tp)
    if period == "custom" or str(period).startswith("custom:"):
        period_label = period if str(period).startswith("custom:") else f"custom:{sample_days}"
    else:
        period_label = str(period)
    miss = missing or []
    return {
        "ok": True,
        "exchange": EXCHANGE,
        "symbol": symbol_key,
        "symbol_label": SYMBOLS[symbol_key]["label"],
        "start_hour": start_hour,
        "end_hour": END_HOUR,
        "period": period_label,
        "sample_days_requested": sample_days,
        "straddle_premium": prem,
        "take_profit": tp,
        "weekend_filter": we_mode,
        "timeframe": TIMEFRAME,
        "price_source": price_source,
        "inst_id": inst_id,
        "timezone": "Asia/Shanghai",
        "rows_all": rows_all,
        "rows": rows,
        "summary": summary,
        "missing_days": miss[:30],
        "missing_count": len(miss),
    }


def reframe_amp_stats(
    *,
    rows_all: list[dict[str, Any]],
    symbol: str = "eth",
    start_hour: int = 16,
    period: str = "2m",
    sample_days: int = 60,
    straddle_premium: Any = None,
    take_profit: Any = None,
    weekend_filter: Any = "all",
    price_source: str = "",
    inst_id: str = "",
    missing: Optional[list[str]] = None,
) -> dict[str, Any]:
    """已有日表上改周末/权利金/止盈,不拉 K 线."""
    key = normalize_symbol(symbol)
    return build_amp_result(
        rows_all=list(rows_all or []),
        symbol_key=key,
        start_hour=int(start_hour),
        period=period,
        sample_days=int(sample_days or 60),
        straddle_premium=straddle_premium,
        take_profit=take_profit,
        weekend_filter=weekend_filter,
        price_source=price_source,
        inst_id=inst_id,
        missing=missing,
    )


def rows_page(rows: list[dict[str, Any]], *, page: int = 1, page_size: int = 20) -> dict[str, Any]:
    page = max(1, int(page or 1))
    page_size = max(5, min(100, int(page_size or 20)))
    total = len(rows)
    start = (page - 1) * page_size
    chunk = rows[start : start + page_size]
    return {
        "page": page,
        "page_size": page_size,
        "total": total,
        "total_pages": max(1, (total + page_size - 1) // page_size) if total else 1,
        "rows": chunk,
    }


def build_export_csv(payload: dict[str, Any]) -> str:
    buf = io.StringIO()
    # Excel 友好 BOM
    buf.write("\ufeff")
    w = csv.writer(buf)
    s = payload.get("summary") or {}
    w.writerow(["【统计摘要】"])
    w.writerow(["交易所", payload.get("exchange")])
    w.writerow(["标的", payload.get("symbol_label")])
    w.writerow(["价源", payload.get("price_source")])
    w.writerow(["起点整点", f"{payload.get('start_hour')}:00"])
    w.writerow(["终点", f"{payload.get('end_hour')}:00"])
    w.writerow(["周期", payload.get("period")])
    w.writerow(["周末筛选", payload.get("weekend_filter")])
    w.writerow(["样本数", s.get("sample_count")])
    w.writerow(["最大振幅", s.get("max_amplitude"), "日期", s.get("max_amplitude_day")])
    w.writerow(["振幅均值", s.get("avg_amplitude"), "中位数", s.get("median_amplitude")])
    w.writerow(["开→高最大", s.get("max_up_points"), "均值", s.get("avg_up_points")])
    w.writerow(["开→低最大", s.get("max_down_points"), "均值", s.get("avg_down_points")])
    w.writerow(["上涨窗占比", s.get("up_day_ratio"), "下跌窗占比", s.get("down_day_ratio")])
    st = s.get("straddle") or {}
    if st:
        w.writerow([])
        w.writerow(["【买跨对照·双边权利金】", st.get("premium"), "止盈点", st.get("take_profit")])
        w.writerow(["开→高超过权利金", st.get("up_exceed_days"), "占比", st.get("up_exceed_ratio")])
        w.writerow(["开→低超过权利金", st.get("down_exceed_days"), "占比", st.get("down_exceed_ratio")])
        w.writerow(["|涨跌|超过权利金", st.get("abs_change_exceed_days"), "占比", st.get("abs_change_exceed_ratio")])
        if st.get("take_profit") is not None:
            w.writerow(["触达止盈天数", st.get("tp_hit_days"), "占比", st.get("tp_hit_ratio")])
        w.writerow(
            [
                "买跨点数盈亏合计",
                st.get("pnl_total"),
                "日均",
                st.get("pnl_avg"),
                "赚钱天数",
                st.get("win_days"),
                "胜率",
                st.get("win_ratio"),
            ]
        )
        w.writerow(["单日最大赚", st.get("pnl_max"), "单日最大亏", st.get("pnl_min")])
    w.writerow([])
    w.writerow(["【日表明细】"])
    w.writerow(
        [
            "结算日",
            "星期",
            "周末",
            "窗起点",
            "窗终点",
            "开盘",
            "最高",
            "最低",
            "收盘",
            "开→高",
            "开→低",
            "振幅",
            "涨跌值",
            "有效波动",
            "触达止盈",
            "收益",
        ]
    )
    for r in payload.get("rows") or []:
        w.writerow(
            [
                r.get("settlement_day"),
                r.get("weekday_label") or "",
                "是" if r.get("is_weekend") else "否",
                r.get("window_start"),
                r.get("window_end"),
                r.get("open"),
                r.get("high"),
                r.get("low"),
                r.get("close"),
                r.get("up_points"),
                r.get("down_points"),
                r.get("amplitude"),
                r.get("change"),
                r.get("effective_move"),
                "是" if r.get("take_profit_hit") else "否",
                r.get("profit"),
            ]
        )
    return buf.getvalue()


def export_filename(payload: dict[str, Any]) -> str:
    sym = (payload.get("symbol") or "eth").lower()
    sh = int(payload.get("start_hour") or 16)
    period = str(payload.get("period") or "2m").replace(":", "")
    day = datetime.now(APP_TZ).strftime("%Y%m%d")
    return f"okx_{sym}_amp_{sh}to16_{period}_{day}.csv"