"""中控振幅统计:OKX 指数(可降级永续)按时段切窗,点数口径. 仅只读行情;不触及下单链路. """ from __future__ import annotations import csv import io import statistics import time 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, *, span_days: int = 1, ) -> tuple[datetime, datetime]: """返回 [start, end) 的本地时刻;end 为结算日 16:00. span_days=1: 与现口径相同(如 26日16:00→27日16:00) span_days=2: 再往前推 1 天(如 25日16:00→27日16:00) """ if not (0 <= int(start_hour) <= 23): raise ValueError("起点须为 0-23 整点") span = max(1, int(span_days or 1)) 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) if span > 1: start = start - timedelta(days=span - 1) 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 _ohlc_window_metrics( start: datetime, end: datetime, bar_map: dict[int, dict[str, float]], ) -> Optional[dict[str, Any]]: """在 [start, end) 上算开高低收与开→高/开→低/振幅/涨跌.""" 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 return { "window_start": start.strftime("%Y-%m-%d %H:%M"), "window_end": end.strftime("%Y-%m-%d %H:%M"), "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 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, span_days=1) m1 = _ohlc_window_metrics(start, end, bar_map) if m1 is None: return None start2, end2 = window_bounds_for_settlement(settlement, start_hour, span_days=2) m2 = _ohlc_window_metrics(start2, end2, bar_map) wd = settlement.weekday() # Mon=0 … Sun=6 is_we = wd >= 5 row: dict[str, Any] = { "settlement_day": settlement.isoformat(), "weekday": wd, "weekday_label": "六" if wd == 5 else ("日" if wd == 6 else ""), "is_weekend": is_we, **m1, } if m2 is None: row.update( { "window2_start": start2.strftime("%Y-%m-%d %H:%M"), "window2_end": end2.strftime("%Y-%m-%d %H:%M"), "open_2d": None, "high_2d": None, "low_2d": None, "close_2d": None, "up_points_2d": None, "down_points_2d": None, "amplitude_2d": None, "change_2d": None, } ) else: row.update( { "window2_start": m2["window_start"], "window2_end": m2["window_end"], "open_2d": m2["open"], "high_2d": m2["high"], "low_2d": m2["low"], "close_2d": m2["close"], "up_points_2d": m2["up_points"], "down_points_2d": m2["down_points"], "amplitude_2d": m2["amplitude"], "change_2d": m2["change"], } ) return row def normalize_move_points(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 _ensure_weekend_flags(item: dict[str, Any]) -> None: if "is_weekend" in item: return if not item.get("settlement_day"): item.setdefault("weekday_label", "") item.setdefault("is_weekend", False) return 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) def enrich_rows( rows: list[dict[str, Any]], *, move_points: Any = None, ) -> list[dict[str, Any]]: """为日表附加周末标注,以及相对波动点数的两边达标.""" mp = normalize_move_points(move_points) out: list[dict[str, Any]] = [] for r in rows or []: item = dict(r) _ensure_weekend_flags(item) up = float(item.get("up_points") or 0) down = float(item.get("down_points") or 0) amp = float(item.get("amplitude") or 0) hit_up = bool(mp is not None and up >= mp) hit_down = bool(mp is not None and down >= mp) amp_hit = bool(mp is not None and amp >= mp) amp2 = item.get("amplitude_2d") amp2_v = float(amp2) if amp2 is not None and amp2 != "" else None amp_hit_2d = bool(mp is not None and amp2_v is not None and amp2_v >= mp) item["move_points"] = mp item["hit_up"] = hit_up item["hit_down"] = hit_down item["hit_either"] = hit_up or hit_down item["hit_both"] = hit_up and hit_down item["amp_hit"] = amp_hit item["amp_hit_2d"] = amp_hit_2d out.append(item) return out # 兼容旧调用名 def enrich_rows_pnl(rows: list[dict[str, Any]], **kwargs: Any) -> list[dict[str, Any]]: return enrich_rows(rows, move_points=kwargs.get("move_points")) def move_points_stats(rows: list[dict[str, Any]], move_points: float) -> dict[str, Any]: """波动点数达标汇总:开→高/开→低两边.""" mp = float(move_points) if mp <= 0: raise ValueError("波动点数须 > 0") work = enrich_rows(rows, move_points=mp) n = len(work) empty = { "move_points": round(mp, 4), "sample_count": n, "up_hit_days": 0, "up_hit_ratio": None, "down_hit_days": 0, "down_hit_ratio": None, "either_hit_days": 0, "either_hit_ratio": None, "both_hit_days": 0, "both_hit_ratio": None, "amp_hit_days": 0, "amp_hit_ratio": None, "amp_2d_hit_days": 0, "amp_2d_hit_ratio": None, "abs_change_hit_days": 0, "abs_change_hit_ratio": None, } if n <= 0: return empty up_hit = sum(1 for r in work if r.get("hit_up")) down_hit = sum(1 for r in work if r.get("hit_down")) either = sum(1 for r in work if r.get("hit_either")) both = sum(1 for r in work if r.get("hit_both")) amp_hit = sum(1 for r in work if r.get("amp_hit")) amp2_rows = [r for r in work if r.get("amplitude_2d") is not None] amp2_hit = sum(1 for r in work if r.get("amp_hit_2d")) n2 = len(amp2_rows) abs_hit = sum(1 for r in work if abs(float(r.get("change") or 0)) >= mp) empty.update( { "up_hit_days": up_hit, "up_hit_ratio": round(up_hit / n, 4), "down_hit_days": down_hit, "down_hit_ratio": round(down_hit / n, 4), "either_hit_days": either, "either_hit_ratio": round(either / n, 4), "both_hit_days": both, "both_hit_ratio": round(both / n, 4), "amp_hit_days": amp_hit, "amp_hit_ratio": round(amp_hit / n, 4), "amp_2d_hit_days": amp2_hit, "amp_2d_hit_ratio": round(amp2_hit / n2, 4) if n2 else None, "abs_change_hit_days": abs_hit, "abs_change_hit_ratio": round(abs_hit / n, 4), } ) return empty def summarize_rows( rows: list[dict[str, Any]], *, move_points: Any = None, ) -> dict[str, Any]: mp = normalize_move_points(move_points) empty_2d = { "max_amplitude_2d": None, "max_amplitude_2d_day": None, "avg_amplitude_2d": None, "median_amplitude_2d": None, } 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, **empty_2d, "move_points_stats": None, } if mp is not None: out["move_points_stats"] = move_points_stats([], mp) 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) amps2 = [float(r["amplitude_2d"]) for r in rows if r.get("amplitude_2d") is not None] 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), **empty_2d, "move_points_stats": None, } if amps2: max_a2 = max(amps2) out["max_amplitude_2d"] = round(max_a2, 4) out["max_amplitude_2d_day"] = next( r["settlement_day"] for r in rows if r.get("amplitude_2d") is not None and float(r["amplitude_2d"]) == max_a2 ) out["avg_amplitude_2d"] = round(statistics.fmean(amps2), 4) out["median_amplitude_2d"] = round(statistics.median(amps2), 4) if mp is not None: out["move_points_stats"] = move_points_stats(rows, mp) 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 _okx_get_json( client: httpx.Client, url: str, params: dict[str, str], *, retries: int = 8, ) -> dict[str, Any]: """GET OKX 公共行情;遇 429 指数退避重试.""" last_err: Optional[BaseException] = None for attempt in range(max(1, int(retries))): try: r = client.get(url, params=params) if r.status_code == 429: wait = min(12.0, 0.7 * (2**attempt)) time.sleep(wait) last_err = httpx.HTTPStatusError( f"429 Too Many Requests for url '{r.url}'", request=r.request, response=r, ) continue r.raise_for_status() body = r.json() if not isinstance(body, dict): raise RuntimeError("OKX 返回非对象 JSON") return body except httpx.HTTPStatusError as exc: status = exc.response.status_code if exc.response is not None else None if status == 429 and attempt + 1 < retries: wait = min(12.0, 0.7 * (2**attempt)) time.sleep(wait) last_err = exc continue raise except httpx.TransportError as exc: if attempt + 1 < retries: time.sleep(min(8.0, 0.5 * (2**attempt))) last_err = exc continue raise if last_err is not None: raise last_err raise RuntimeError("OKX 请求失败") 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, page_pause_sec: float = 0.12, history_page_pause_sec: float = 0.22, ) -> list[dict[str, Any]]: """拉取 [since_ms, until_ms] 覆盖的 K 线(含边界). OKX 近期接口约仅 1440 根;更早需 history_* 端点续拉. 分页带间隔,429 自动退避重试. """ 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 for page_i in range(max(20, int(max_pages))): if page_i > 0: pause = history_page_pause_sec if switched_history or "history" in active_url else page_pause_sec if pause > 0: time.sleep(pause) params: dict[str, str] = {"instId": inst_id, "bar": bar, "limit": "100"} if after: params["after"] = after body = _okx_get_json(client, active_url, params) 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: # 近期接口到头 → 切历史端点再试 if history_url and not switched_history and after is not None: active_url = history_url switched_history = True time.sleep(max(history_page_pause_sec, 0.35)) continue break oldest_ts = 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) 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 time.sleep(max(history_page_pause_sec, 0.35)) 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 time.sleep(max(history_page_pause_sec, 0.35)) 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"] index_err: Optional[BaseException] = None 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 as exc: index_err = exc # 指数侧已触发限频时先冷却,再降级永续,避免连环 429 time.sleep(1.2) try: 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, ) except Exception as exc: detail = f"index={index_err}; swap={exc}" if index_err else str(exc) raise RuntimeError(f"OKX K线拉取失败({detail})") from exc if not bars: detail = f"index={index_err}" if index_err else "empty" raise RuntimeError(f"OKX 指数与永续 K 线均无数据({detail})") 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, move_points: 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 整点") mp = normalize_move_points(move_points) 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("无可用结算日") # 最远窗起点(含两日振幅,多拉 1 天) oldest = settlements[-1] newest = settlements[0] start0, _ = window_bounds_for_settlement(oldest, sh, span_days=2) _, end1 = window_bounds_for_settlement(newest, sh, span_days=1) 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, move_points=mp, 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, move_points: Any = None, weekend_filter: Any = "all", price_source: str = "", inst_id: str = "", missing: Optional[list[str]] = None, ) -> dict[str, Any]: mp = normalize_move_points(move_points) we_mode = normalize_weekend_filter(weekend_filter) filtered = filter_weekend_rows(rows_all, we_mode) rows = enrich_rows(filtered, move_points=mp) summary = summarize_rows(rows, move_points=mp) 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, "move_points": mp, "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, move_points: 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), move_points=move_points, 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_amplitude_2d"), "日期", s.get("max_amplitude_2d_day")]) w.writerow(["两日振幅均值", s.get("avg_amplitude_2d"), "中位数", s.get("median_amplitude_2d")]) 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")]) mp = s.get("move_points_stats") or {} if mp: w.writerow([]) w.writerow(["【波动点数·振幅占比】", mp.get("move_points")]) w.writerow(["振幅≥点数天数", mp.get("amp_hit_days"), "占比", mp.get("amp_hit_ratio")]) w.writerow(["两日振幅≥点数天数", mp.get("amp_2d_hit_days"), "占比", mp.get("amp_2d_hit_ratio")]) w.writerow(["开→高≥点数天数", mp.get("up_hit_days"), "占比", mp.get("up_hit_ratio")]) w.writerow(["开→低≥点数天数", mp.get("down_hit_days"), "占比", mp.get("down_hit_ratio")]) w.writerow(["|涨跌|≥点数天数", mp.get("abs_change_hit_days"), "占比", mp.get("abs_change_hit_ratio")]) 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("window2_start"), r.get("window2_end"), r.get("amplitude_2d"), r.get("up_points_2d"), r.get("down_points_2d"), r.get("move_points") if r.get("move_points") is not None else "", "是" if r.get("amp_hit") else ("否" if r.get("move_points") is not None else ""), "是" if r.get("amp_hit_2d") else ("否" if r.get("move_points") is not None and r.get("amplitude_2d") is not None else ""), ] ) 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"