dekun/crypto_monitor
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Sync OKX options closed trades into hub archive with a separate tab.

Browse Source 
Mirror perpetual archive flow into archive_options_trade_cache for offline calendar and review.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
dekun
2026-07-24 00:39:59 +08:00
parent 6f1ae14b3d
commit 54f1857fa2
8 changed files with 1058 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/hub/hub_bridge.py
+66
View File
@@ -672,6 +672,72 @@ def register_hub_routes(app):
}
)
@app.route("/api/hub/options/review/archive")
@_hub_auth_required
def api_hub_options_review_archive():
"""中控期权档案:近 N 天已平仓复盘记录(默认排除对冲腿)."""
from datetime import datetime, timedelta
from zoneinfo import ZoneInfo
from flask import current_app
from lib.options.options_review_lib import (
compute_review_stats,
ensure_local_review_synced,
list_review_trades,
)
c = _ctx()
get_db = c.get("get_db")
if not get_db:
return jsonify({"ok": False, "msg": "HUB_CTX 缺少 get_db"}), 500
try:
days = int(request.args.get("days") or "365")
except ValueError:
days = 365
days = max(1, min(days, 3650))
try:
limit = int(request.args.get("limit") or "2000")
except ValueError:
limit = 2000
limit = max(1, min(limit, 5000))
include_hedge_legs = str(request.args.get("include_hedge_legs") or "").strip() in (
"1",
"true",
"yes",
)
tz = ZoneInfo("Asia/Shanghai")
closed_from = (datetime.now(tz) - timedelta(days=days)).strftime("%Y-%m-%d")
cfg = (current_app.extensions or {}).get("options_cfg") or {}
ex = cfg.get("exchange_options")
conn = get_db()
try:
ensure_local_review_synced(conn, ex=ex, backfill_exchange_pnl=bool(ex))
trades = list_review_trades(
conn,
include_hedge_legs=include_hedge_legs,
closed_from=closed_from,
limit=limit,
offset=0,
)
stats = compute_review_stats(
conn,
include_hedge_legs=include_hedge_legs,
closed_from=closed_from,
)
finally:
conn.close()
return jsonify(
{
"ok": True,
"days": days,
"limit": limit,
"product": "options",
"trades": trades,
"stats": stats,
}
)
@app.route("/api/hub/trades/today")
@_hub_auth_required
def api_hub_trades_today():
lib/hub/hub_options_archive_lib.py
+599
View File
@@ -0,0 +1,599 @@
"""中控期权档案:同步 OKX options_review_trades 到 hub_symbol_archive.db."""
from __future__ import annotations
import json
import time
from pathlib import Path
from typing import Any
from lib.hub.hub_symbol_archive_lib import (
TRADING_DAY_RESET_HOUR,
_connect,
default_db_path,
init_db as init_perp_archive_db,
ms_to_trading_day,
parse_wall_clock_ms,
resolve_period_bounds,
trading_day_bounds_ms,
)
def _now_ms() -> int:
return int(time.time() * 1000)
def init_options_archive_db(db_path: Path | None = None) -> None:
"""确保期权缓存表存在(与永续共用同一 SQLite)."""
init_perp_archive_db(db_path)
conn = _connect(db_path)
try:
conn.execute(
"""
CREATE TABLE IF NOT EXISTS archive_options_trade_cache (
exchange_key TEXT NOT NULL,
history_key TEXT NOT NULL,
source_type TEXT,
underlying TEXT,
opened_at TEXT,
closed_at TEXT,
opened_at_ms INTEGER,
closed_at_ms INTEGER,
hold_seconds INTEGER,
realized_pnl_total REAL,
status_raw TEXT,
pos_id TEXT,
inst_id TEXT,
opt_type TEXT,
strike REAL,
exp_time TEXT,
sheets INTEGER,
open_avg REAL,
close_avg REAL,
premium_paid REAL,
realized_pnl REAL,
hedge_plan_id INTEGER,
plan_close_reason TEXT,
realized_pnl_perp REAL,
realized_pnl_options REAL,
premium_total REAL,
direction TEXT,
tp REAL,
sl REAL,
target_price REAL,
target_price_up REAL,
target_price_down REAL,
legs_json TEXT,
linked_hedge_plan_id INTEGER,
excluded_as_hedge_leg INTEGER DEFAULT 0,
strategy_tag TEXT,
result_tag TEXT,
reviewed INTEGER DEFAULT 0,
source_label TEXT,
payload_json TEXT NOT NULL,
synced_at INTEGER NOT NULL,
PRIMARY KEY (exchange_key, history_key)
)
"""
)
conn.execute(
"""
CREATE INDEX IF NOT EXISTS idx_archive_options_closed
ON archive_options_trade_cache (exchange_key, closed_at_ms)
"""
)
finally:
conn.close()
def purge_stale_options_trades_cache(
exchange_key: str,
active_history_keys: list[str],
*,
db_path: Path | None = None,
) -> int:
init_options_archive_db(db_path)
ex_k = (exchange_key or "").strip().lower()
if not ex_k:
return 0
active = {str(k).strip() for k in (active_history_keys or []) if str(k).strip()}
conn = _connect(db_path)
try:
rows = conn.execute(
"SELECT history_key FROM archive_options_trade_cache WHERE exchange_key=?",
(ex_k,),
).fetchall()
stale = [r["history_key"] for r in rows if r["history_key"] not in active]
removed = 0
for hk in stale:
cur = conn.execute(
"DELETE FROM archive_options_trade_cache WHERE exchange_key=? AND history_key=?",
(ex_k, hk),
)
removed += int(cur.rowcount or 0)
return removed
finally:
conn.close()
def _optional_float(raw: Any) -> float | None:
if raw in (None, ""):
return None
try:
return float(raw)
except (TypeError, ValueError):
return None
def _optional_int(raw: Any) -> int | None:
if raw in (None, ""):
return None
try:
return int(raw)
except (TypeError, ValueError):
return None
def upsert_options_trades_cache(
exchange_key: str,
trades: list[dict[str, Any]],
*,
db_path: Path | None = None,
prune_missing: bool = True,
) -> dict[str, int]:
init_options_archive_db(db_path)
ex_k = (exchange_key or "").strip().lower()
if not ex_k:
return {"upserted": 0, "removed": 0}
now = _now_ms()
n = 0
active_keys: list[str] = []
conn = _connect(db_path)
try:
for t in trades or []:
if not isinstance(t, dict):
continue
hk = str(t.get("history_key") or "").strip()
if not hk:
continue
if int(t.get("excluded_as_hedge_leg") or 0):
continue
active_keys.append(hk)
opened_at = t.get("opened_at")
closed_at = t.get("closed_at")
opened_ms = t.get("opened_at_ms") or parse_wall_clock_ms(opened_at)
closed_ms = t.get("closed_at_ms") or parse_wall_clock_ms(closed_at)
entry = t.get("entry") if isinstance(t.get("entry"), dict) else {}
strategy_tag = t.get("strategy_tag") or (entry or {}).get("strategy_tag")
result_tag = t.get("result_tag") or (entry or {}).get("result_tag")
reviewed = 1 if t.get("reviewed") or entry else 0
row = dict(t)
row["exchange_key"] = ex_k
payload = json.dumps(row, ensure_ascii=False, default=str)
conn.execute(
"""
INSERT INTO archive_options_trade_cache (
exchange_key, history_key, source_type, underlying,
opened_at, closed_at, opened_at_ms, closed_at_ms, hold_seconds,
realized_pnl_total, status_raw,
pos_id, inst_id, opt_type, strike, exp_time, sheets,
open_avg, close_avg, premium_paid, realized_pnl,
hedge_plan_id, plan_close_reason, realized_pnl_perp, realized_pnl_options,
premium_total, direction, tp, sl, target_price, target_price_up, target_price_down,
legs_json, linked_hedge_plan_id, excluded_as_hedge_leg,
strategy_tag, result_tag, reviewed, source_label,
payload_json, synced_at
) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)
ON CONFLICT(exchange_key, history_key) DO UPDATE SET
source_type=excluded.source_type,
underlying=excluded.underlying,
opened_at=excluded.opened_at,
closed_at=excluded.closed_at,
opened_at_ms=excluded.opened_at_ms,
closed_at_ms=excluded.closed_at_ms,
hold_seconds=excluded.hold_seconds,
realized_pnl_total=excluded.realized_pnl_total,
status_raw=excluded.status_raw,
pos_id=excluded.pos_id,
inst_id=excluded.inst_id,
opt_type=excluded.opt_type,
strike=excluded.strike,
exp_time=excluded.exp_time,
sheets=excluded.sheets,
open_avg=excluded.open_avg,
close_avg=excluded.close_avg,
premium_paid=excluded.premium_paid,
realized_pnl=excluded.realized_pnl,
hedge_plan_id=excluded.hedge_plan_id,
plan_close_reason=excluded.plan_close_reason,
realized_pnl_perp=excluded.realized_pnl_perp,
realized_pnl_options=excluded.realized_pnl_options,
premium_total=excluded.premium_total,
direction=excluded.direction,
tp=excluded.tp,
sl=excluded.sl,
target_price=excluded.target_price,
target_price_up=excluded.target_price_up,
target_price_down=excluded.target_price_down,
legs_json=excluded.legs_json,
linked_hedge_plan_id=excluded.linked_hedge_plan_id,
excluded_as_hedge_leg=excluded.excluded_as_hedge_leg,
strategy_tag=excluded.strategy_tag,
result_tag=excluded.result_tag,
reviewed=excluded.reviewed,
source_label=excluded.source_label,
payload_json=excluded.payload_json,
synced_at=excluded.synced_at
""",
(
ex_k,
hk,
t.get("source_type"),
t.get("underlying"),
opened_at,
closed_at,
int(opened_ms) if opened_ms else None,
int(closed_ms) if closed_ms else None,
_optional_int(t.get("hold_seconds")),
float(t.get("realized_pnl_total") or t.get("realized_pnl") or 0),
t.get("status_raw"),
t.get("pos_id"),
t.get("inst_id"),
t.get("opt_type"),
_optional_float(t.get("strike")),
t.get("exp_time"),
_optional_int(t.get("sheets")),
_optional_float(t.get("open_avg")),
_optional_float(t.get("close_avg")),
_optional_float(t.get("premium_paid")),
_optional_float(t.get("realized_pnl")),
_optional_int(t.get("hedge_plan_id")),
t.get("plan_close_reason"),
_optional_float(t.get("realized_pnl_perp")),
_optional_float(t.get("realized_pnl_options")),
_optional_float(t.get("premium_total")),
t.get("direction"),
_optional_float(t.get("tp")),
_optional_float(t.get("sl")),
_optional_float(t.get("target_price")),
_optional_float(t.get("target_price_up")),
_optional_float(t.get("target_price_down")),
t.get("legs_json")
if isinstance(t.get("legs_json"), str)
else (json.dumps(t.get("legs"), ensure_ascii=False) if t.get("legs") else None),
_optional_int(t.get("linked_hedge_plan_id")),
int(t.get("excluded_as_hedge_leg") or 0),
strategy_tag,
result_tag,
reviewed,
t.get("source_label"),
payload,
now,
),
)
n += 1
finally:
conn.close()
removed = 0
if prune_missing:
removed = purge_stale_options_trades_cache(ex_k, active_keys, db_path=db_path)
return {"upserted": n, "removed": removed}
def _options_row_to_dict(row: Any) -> dict[str, Any]:
out: dict[str, Any] = dict(row)
payload = {}
raw = out.get("payload_json")
if raw:
try:
payload = json.loads(raw) if isinstance(raw, str) else {}
except (TypeError, ValueError, json.JSONDecodeError):
payload = {}
if isinstance(payload, dict):
for k, v in payload.items():
if k not in out or out.get(k) in (None, ""):
out[k] = v
pnl = float(out.get("realized_pnl_total") or out.get("realized_pnl") or 0)
out["realized_pnl_total"] = pnl
out["pnl_amount"] = pnl # 复用永续统计/日历字段名
hold_sec = out.get("hold_seconds")
if hold_sec is not None:
try:
out["hold_minutes"] = round(float(hold_sec) / 60.0, 2)
except (TypeError, ValueError):
pass
if not out.get("opened_at_ms") and out.get("opened_at"):
ms = parse_wall_clock_ms(out.get("opened_at"))
if ms:
out["opened_at_ms"] = int(ms)
if not out.get("closed_at_ms") and out.get("closed_at"):
ms = parse_wall_clock_ms(out.get("closed_at"))
if ms:
out["closed_at_ms"] = int(ms)
out["trade_id"] = out.get("history_key")
out["id"] = out.get("history_key")
out["symbol"] = out.get("inst_id") or out.get("underlying") or ""
return out
def _empty_options_stats() -> dict[str, Any]:
return {
"open_count": 0,
"sick_count": 0,
"sick_pct": 0.0,
"pnl_total": 0.0,
"pnl_ex_sick": 0.0,
"win_count": 0,
"loss_count": 0,
"avg_win": 0.0,
"avg_loss": 0.0,
"max_win": 0.0,
"max_loss": 0.0,
"win_rate": 0.0,
"profit_loss_ratio": 0.0,
"turnover_total": 0.0,
"commission_total": 0.0,
"premium_total": 0.0,
"by_exchange": {},
"by_source_type": {},
}
def _compute_options_period_stats(trade_rows: list[dict[str, Any]]) -> dict[str, Any]:
st = _empty_options_stats()
wins: list[float] = []
losses: list[float] = []
by_ex: dict[str, dict[str, Any]] = {}
by_src: dict[str, dict[str, Any]] = {}
def bucket() -> dict[str, Any]:
return {
"open_count": 0,
"pnl_total": 0.0,
"win_count": 0,
"loss_count": 0,
"premium_total": 0.0,
}
for td in trade_rows:
pnl = float(td.get("pnl_amount") or td.get("realized_pnl_total") or 0)
ex = str(td.get("exchange_key") or "okx")
src = str(td.get("source_type") or td.get("source_label") or "?")
prem = float(td.get("premium_total") or td.get("premium_paid") or 0)
st["open_count"] += 1
st["pnl_total"] += pnl
st["premium_total"] += prem
if pnl > 0.0001:
st["win_count"] += 1
wins.append(pnl)
elif pnl < -0.0001:
st["loss_count"] += 1
losses.append(pnl)
if ex not in by_ex:
by_ex[ex] = bucket()
by_ex[ex]["open_count"] += 1
by_ex[ex]["pnl_total"] += pnl
by_ex[ex]["premium_total"] += prem
if pnl > 0.0001:
by_ex[ex]["win_count"] += 1
elif pnl < -0.0001:
by_ex[ex]["loss_count"] += 1
if src not in by_src:
by_src[src] = bucket()
by_src[src]["open_count"] += 1
by_src[src]["pnl_total"] += pnl
total = int(st["open_count"] or 0)
st["pnl_ex_sick"] = round(float(st["pnl_total"]), 4)
st["pnl_total"] = round(float(st["pnl_total"]), 4)
st["premium_total"] = round(float(st["premium_total"]), 4)
st["avg_win"] = round(sum(wins) / len(wins), 4) if wins else 0.0
st["avg_loss"] = round(sum(losses) / len(losses), 4) if losses else 0.0
st["max_win"] = round(max(wins), 4) if wins else 0.0
st["max_loss"] = round(min(losses), 4) if losses else 0.0
st["win_rate"] = round(st["win_count"] / total * 100, 1) if total else 0.0
if wins and losses and abs(st["avg_loss"]) > 1e-9:
st["profit_loss_ratio"] = round(abs(st["avg_win"] / st["avg_loss"]), 2)
for ex, b in by_ex.items():
b["pnl_total"] = round(float(b["pnl_total"]), 4)
b["premium_total"] = round(float(b["premium_total"]), 4)
b["sick_count"] = 0
b["sick_pct"] = 0.0
b["pnl_ex_sick"] = b["pnl_total"]
b["avg_win"] = 0.0
b["avg_loss"] = 0.0
b["max_win"] = 0.0
b["max_loss"] = 0.0
b["win_rate"] = (
round(b["win_count"] / b["open_count"] * 100, 1) if b["open_count"] else 0.0
)
b["profit_loss_ratio"] = 0.0
b["turnover_total"] = 0.0
b["commission_total"] = 0.0
for src, b in by_src.items():
b["pnl_total"] = round(float(b["pnl_total"]), 4)
st["by_exchange"] = by_ex
st["by_source_type"] = by_src
return st
def list_daily_options_trades(
trading_day: str = "",
*,
period: str = "",
date_from: str = "",
date_to: str = "",
exchange_key: str = "",
filter_profit: bool = False,
filter_loss: bool = False,
search: str = "",
source_type: str = "",
db_path: Path | None = None,
) -> dict[str, Any]:
init_options_archive_db(db_path)
p = (period or "today").strip().lower() or "today"
start_ms, end_ms, df, dt, period_label = resolve_period_bounds(
period=p,
trading_day=trading_day,
date_from=date_from,
date_to=date_to,
)
ex_filter = (exchange_key or "").strip().lower()
src_filter = (source_type or "").strip().lower()
conn = _connect(db_path)
try:
params: list[Any] = [start_ms, end_ms]
where = "closed_at_ms IS NOT NULL AND closed_at_ms >= ? AND closed_at_ms < ?"
where += " AND COALESCE(excluded_as_hedge_leg,0)=0"
if ex_filter:
where += " AND exchange_key=?"
params.append(ex_filter)
if src_filter:
where += " AND LOWER(COALESCE(source_type,''))=?"
params.append(src_filter)
rows = conn.execute(
f"""
SELECT * FROM archive_options_trade_cache
WHERE {where}
ORDER BY closed_at_ms DESC, history_key DESC
""",
params,
).fetchall()
trades: list[dict[str, Any]] = []
q = (search or "").strip().lower()
for r in rows:
td = _options_row_to_dict(r)
pnl = float(td.get("pnl_amount") or 0)
if filter_profit and pnl <= 0.0001:
continue
if filter_loss and pnl >= -0.0001:
continue
if q:
blob = " ".join(
str(td.get(k) or "")
for k in (
"underlying",
"inst_id",
"exchange_key",
"source_type",
"source_label",
"opt_type",
"strategy_tag",
"result_tag",
"direction",
)
).lower()
if q not in blob:
continue
trades.append(td)
return {
"period": p,
"period_label": period_label,
"trading_day": dt,
"date_from": df,
"date_to": dt,
"product": "options",
"trades": trades,
"stats": _compute_options_period_stats(trades),
}
finally:
conn.close()
def list_archive_options_calendar(
year: int,
month: int,
*,
exchange_key: str = "",
db_path: Path | None = None,
reset_hour: int = TRADING_DAY_RESET_HOUR,
) -> dict[str, Any]:
init_options_archive_db(db_path)
y = int(year)
m = int(month)
if m < 1 or m > 12:
raise ValueError("month 无效")
from datetime import datetime, timedelta
first = f"{y:04d}-{m:02d}-01"
if m == 12:
next_first = datetime(y + 1, 1, 1)
else:
next_first = datetime(y, m + 1, 1)
last = (next_first - timedelta(days=1)).strftime("%Y-%m-%d")
start_ms, _ = trading_day_bounds_ms(first, reset_hour=reset_hour)
_, end_ms = trading_day_bounds_ms(last, reset_hour=reset_hour)
ex_filter = (exchange_key or "").strip().lower()
conn = _connect(db_path)
try:
params: list[Any] = [start_ms, end_ms]
where = (
"closed_at_ms IS NOT NULL AND closed_at_ms >= ? AND closed_at_ms < ?"
" AND COALESCE(excluded_as_hedge_leg,0)=0"
)
if ex_filter:
where += " AND exchange_key=?"
params.append(ex_filter)
rows = conn.execute(
f"SELECT * FROM archive_options_trade_cache WHERE {where}",
params,
).fetchall()
days: dict[str, dict[str, Any]] = {}
for r in rows:
td = _options_row_to_dict(r)
closed_ms = td.get("closed_at_ms") or parse_wall_clock_ms(td.get("closed_at"))
if not closed_ms:
continue
day = ms_to_trading_day(int(closed_ms), reset_hour=reset_hour)
if not day or day < first or day > last:
continue
bucket = days.setdefault(
day,
{
"trading_day": day,
"open_count": 0,
"sick_count": 0,
"pnl_total": 0.0,
"turnover_total": 0.0,
"commission_total": 0.0,
"has_sick": False,
},
)
bucket["open_count"] += 1
bucket["pnl_total"] += float(td.get("pnl_amount") or 0)
for d in days.values():
d["pnl_total"] = round(float(d["pnl_total"]), 4)
month_pnl = sum(float(d["pnl_total"]) for d in days.values())
month_count = sum(int(d["open_count"]) for d in days.values())
return {
"year": y,
"month": m,
"date_from": first,
"date_to": last,
"product": "options",
"days": days,
"month_pnl_total": round(month_pnl, 4),
"month_open_count": month_count,
}
finally:
conn.close()
def sync_options_exchange_archive(
exchange_key: str,
trades: list[dict[str, Any]],
*,
db_path: Path | None = None,
) -> dict[str, Any]:
"""仅缓存期权交易,不做 K 线."""
r = upsert_options_trades_cache(
exchange_key, trades, db_path=db_path, prune_missing=True
)
return {
"ok": True,
"exchange_key": (exchange_key or "").strip().lower(),
"product": "options",
"trades_upserted": r.get("upserted", 0),
"trades_removed": r.get("removed", 0),
"trade_count": len(trades or []),
}