dekun
538 lines
19 KiB
Python
538 lines
19 KiB
Python
"""期权本地交易统计(胜率 / 盈亏 / 持仓时长)."""
|
||
from __future__ import annotations
|
||
|
||
from datetime import datetime
|
||
from typing import Any
|
||
|
||
from lib.instance.instance_embed_context_lib import profit_loss_ratio_from_averages
|
||
from lib.options.options_db import init_options_tables
|
||
|
||
|
||
def _safe_float(v: Any) -> float | None:
|
||
if v is None or v == "":
|
||
return None
|
||
try:
|
||
return float(v)
|
||
except (TypeError, ValueError):
|
||
return None
|
||
|
||
|
||
def _underlying_index_usdt(ex: Any, underly: str) -> float | None:
|
||
"""取标的 USDT 近似指数(币本位已平盈亏折 U).优先公开 ticker,避免私钥失败."""
|
||
u = (underly or "ETH").strip().upper() or "ETH"
|
||
pubs: list[Any] = []
|
||
try:
|
||
from lib.sim.hooks import _APP_MODULE, _sim_public_exchange
|
||
|
||
pub = _sim_public_exchange(ex) if _APP_MODULE is not None else None
|
||
if pub is not None:
|
||
pubs.append(pub)
|
||
except Exception:
|
||
pass
|
||
if ex is not None and ex not in pubs:
|
||
pubs.append(ex)
|
||
from lib.exchange.okx_options_lib import fetch_index_price
|
||
|
||
for pub in pubs:
|
||
try:
|
||
if hasattr(pub, "public_get_market_ticker"):
|
||
rows = (pub.public_get_market_ticker({"instId": f"{u}-USDT"}) or {}).get("data") or []
|
||
if rows:
|
||
last = _safe_float(rows[0].get("last") or rows[0].get("lastPx"))
|
||
if last is not None and last > 0:
|
||
return float(last)
|
||
except Exception:
|
||
pass
|
||
try:
|
||
px = fetch_index_price(pub, f"{u}-USD")
|
||
if px is not None and float(px) > 0:
|
||
return float(px)
|
||
except Exception:
|
||
pass
|
||
try:
|
||
t = pub.fetch_ticker(f"{u}/USDT") or {}
|
||
last = _safe_float(t.get("last") or t.get("close"))
|
||
if last is not None and last > 0:
|
||
return float(last)
|
||
except Exception:
|
||
continue
|
||
return None
|
||
|
||
|
||
def history_pnl_to_usdt(history: list[dict[str, Any]], ex: Any = None) -> list[dict[str, Any]]:
|
||
"""
|
||
统计用:优先已冻结的 realized_pnl_usdt(平仓当时口径,含手续费后净额);
|
||
否则币本位用 close_index_px(再否则实时指数)折 U;USDC 原样.
|
||
折算失败的币仓剔除盈亏字段,避免把「币数量」当成 U.
|
||
"""
|
||
from lib.options.options_margin_mode_lib import margin_mode_from_inst_id, premium_ccy_for_mode
|
||
|
||
idx_cache: dict[str, float | None] = {}
|
||
out: list[dict[str, Any]] = []
|
||
for row in history:
|
||
r = dict(row)
|
||
if r.get("status") == "open":
|
||
out.append(r)
|
||
continue
|
||
frozen = _safe_float(r.get("realized_pnl_usdt"))
|
||
if frozen is not None:
|
||
r["realized_pnl"] = round(float(frozen), 4)
|
||
out.append(r)
|
||
continue
|
||
pnl = _safe_float(r.get("realized_pnl"))
|
||
if pnl is None:
|
||
out.append(r)
|
||
continue
|
||
inst = str(r.get("inst_id") or "")
|
||
underly = str(r.get("underlying") or (inst.split("-")[0] if inst else "ETH") or "ETH")
|
||
ccy = str(r.get("premium_ccy") or "").strip().upper()
|
||
if not ccy:
|
||
ccy = premium_ccy_for_mode(margin_mode_from_inst_id(inst), underly)
|
||
if ccy in ("ETH", "BTC"):
|
||
idx = _safe_float(r.get("close_index_px"))
|
||
if idx is None or idx <= 0:
|
||
if underly not in idx_cache:
|
||
idx_cache[underly] = _underlying_index_usdt(ex, underly)
|
||
idx = idx_cache.get(underly)
|
||
if idx is None or idx <= 0:
|
||
r["realized_pnl"] = None
|
||
else:
|
||
r["realized_pnl"] = round(float(pnl) * float(idx), 4)
|
||
else:
|
||
r["realized_pnl"] = round(float(pnl), 4)
|
||
out.append(r)
|
||
return out
|
||
|
||
|
||
def _parse_ts(raw: Any) -> datetime | None:
|
||
if raw is None or raw == "":
|
||
return None
|
||
s = str(raw).strip().replace(" ", "T", 1)
|
||
try:
|
||
return datetime.fromisoformat(s)
|
||
except (TypeError, ValueError):
|
||
return None
|
||
|
||
|
||
def _hold_seconds(created_at: Any, closed_at: Any) -> float | None:
|
||
start = _parse_ts(created_at)
|
||
end = _parse_ts(closed_at)
|
||
if start is None or end is None:
|
||
return None
|
||
sec = (end - start).total_seconds()
|
||
return sec if sec >= 0 else None
|
||
|
||
|
||
def _avg_seconds(values: list[float]) -> float | None:
|
||
if not values:
|
||
return None
|
||
return round(sum(values) / len(values), 1)
|
||
|
||
|
||
def compute_options_stats_from_history(history: list[dict[str, Any]]) -> dict[str, Any]:
|
||
"""基于期权历史列表(交易所)计算统计."""
|
||
wins: list[float] = []
|
||
losses: list[float] = []
|
||
win_holds: list[float] = []
|
||
loss_holds: list[float] = []
|
||
all_holds: list[float] = []
|
||
open_holds: list[float] = []
|
||
now = datetime.now()
|
||
|
||
for row in history:
|
||
if row.get("status") == "open":
|
||
start = _parse_ts(row.get("created_at"))
|
||
if start is not None:
|
||
sec = (now - start).total_seconds()
|
||
if sec >= 0:
|
||
open_holds.append(sec)
|
||
continue
|
||
pnl_raw = row.get("realized_pnl")
|
||
if pnl_raw is None:
|
||
continue
|
||
try:
|
||
pnl = float(pnl_raw)
|
||
except (TypeError, ValueError):
|
||
continue
|
||
hold = _hold_seconds(row.get("created_at"), row.get("closed_at"))
|
||
if hold is not None:
|
||
all_holds.append(hold)
|
||
if pnl > 0:
|
||
wins.append(pnl)
|
||
if hold is not None:
|
||
win_holds.append(hold)
|
||
elif pnl < 0:
|
||
losses.append(pnl)
|
||
if hold is not None:
|
||
loss_holds.append(hold)
|
||
|
||
total_closed = len(wins) + len(losses)
|
||
win_rate = round(len(wins) / total_closed * 100, 2) if total_closed else 0
|
||
avg_win = sum(wins) / len(wins) if wins else None
|
||
avg_loss = sum(losses) / len(losses) if losses else None
|
||
|
||
total_profit = round(sum(wins), 4) if wins else 0.0
|
||
total_loss = round(abs(sum(losses)), 4) if losses else 0.0
|
||
net_realized = round(sum(wins) + sum(losses), 4)
|
||
return {
|
||
"total_closed": total_closed,
|
||
"win_count": len(wins),
|
||
"loss_count": len(losses),
|
||
"win_rate": win_rate,
|
||
"profit_loss_ratio": profit_loss_ratio_from_averages(avg_win, avg_loss),
|
||
"avg_win": round(avg_win, 4) if avg_win is not None else None,
|
||
"avg_loss": round(abs(avg_loss), 4) if avg_loss is not None else None,
|
||
"total_profit": total_profit,
|
||
"total_loss": total_loss,
|
||
"net_realized_pnl": net_realized,
|
||
"avg_hold_sec": _avg_seconds(all_holds),
|
||
"avg_win_hold_sec": _avg_seconds(win_holds),
|
||
"avg_loss_hold_sec": _avg_seconds(loss_holds),
|
||
"open_count": len(open_holds),
|
||
"avg_open_hold_sec": _avg_seconds(open_holds),
|
||
}
|
||
|
||
|
||
_STATS_WINDOWS = ("today", "7d", "30d", "all")
|
||
|
||
|
||
def normalize_stats_window(raw: Any) -> str:
|
||
w = str(raw or "7d").strip().lower()
|
||
if w in ("1d", "day", "今日"):
|
||
return "today"
|
||
if w in ("7", "week"):
|
||
return "7d"
|
||
if w in ("30", "month"):
|
||
return "30d"
|
||
if w in ("0", "全部", "total"):
|
||
return "all"
|
||
return w if w in _STATS_WINDOWS else "7d"
|
||
|
||
|
||
def stats_window_bounds(window: str, *, now: datetime | None = None) -> tuple[datetime | None, datetime | None]:
|
||
"""返回闭区间起止(本地/业务时区近似用本地 now).all → (None, None)."""
|
||
w = normalize_stats_window(window)
|
||
end = now or datetime.now()
|
||
if w == "all":
|
||
return None, None
|
||
if w == "today":
|
||
start = end.replace(hour=0, minute=0, second=0, microsecond=0)
|
||
return start, end
|
||
days = 7 if w == "7d" else 30
|
||
from datetime import timedelta
|
||
|
||
start = end - timedelta(days=days)
|
||
return start, end
|
||
|
||
|
||
def filter_history_for_stats_window(
|
||
history: list[dict[str, Any]],
|
||
window: str,
|
||
*,
|
||
now: datetime | None = None,
|
||
) -> list[dict[str, Any]]:
|
||
"""窗口内已平仓 + 当前持仓(open)."""
|
||
start, end = stats_window_bounds(window, now=now)
|
||
out: list[dict[str, Any]] = []
|
||
for row in history or []:
|
||
if row.get("status") == "open":
|
||
out.append(row)
|
||
continue
|
||
if start is None:
|
||
out.append(row)
|
||
continue
|
||
ts = _parse_ts(row.get("closed_at")) or _parse_ts(row.get("created_at"))
|
||
if ts is None:
|
||
continue
|
||
if ts < start:
|
||
continue
|
||
if end is not None and ts > end:
|
||
continue
|
||
out.append(row)
|
||
return out
|
||
|
||
|
||
def compute_options_stats_for_window(
|
||
history: list[dict[str, Any]],
|
||
window: str,
|
||
*,
|
||
now: datetime | None = None,
|
||
) -> dict[str, Any]:
|
||
w = normalize_stats_window(window)
|
||
filtered = filter_history_for_stats_window(history, w, now=now)
|
||
stats = compute_options_stats_from_history(filtered)
|
||
start, end = stats_window_bounds(w, now=now)
|
||
stats["window"] = w
|
||
stats["window_start"] = start.strftime("%Y-%m-%d %H:%M:%S") if start else None
|
||
stats["window_end"] = end.strftime("%Y-%m-%d %H:%M:%S") if end else None
|
||
return stats
|
||
|
||
|
||
def _row_get(row: Any, key: str, default: Any = None) -> Any:
|
||
try:
|
||
if hasattr(row, "keys") and key in row.keys():
|
||
return row[key]
|
||
except Exception:
|
||
pass
|
||
if isinstance(row, dict):
|
||
return row.get(key, default)
|
||
return default
|
||
|
||
|
||
def _closed_trade_dicts(closed_rows) -> list[dict[str, Any]]:
|
||
out: list[dict[str, Any]] = []
|
||
for row in closed_rows or []:
|
||
out.append(
|
||
{
|
||
"status": "closed",
|
||
"realized_pnl": _row_get(row, "realized_pnl"),
|
||
"realized_pnl_usdt": _row_get(row, "realized_pnl_usdt"),
|
||
"close_index_px": _row_get(row, "close_index_px"),
|
||
"created_at": _row_get(row, "created_at"),
|
||
"closed_at": _row_get(row, "closed_at"),
|
||
"inst_id": _row_get(row, "inst_id") or "",
|
||
"underlying": _row_get(row, "underlying") or "",
|
||
"premium_ccy": _row_get(row, "premium_ccy") or "",
|
||
}
|
||
)
|
||
return out
|
||
|
||
|
||
_CLOSED_STATS_SELECT = """
|
||
SELECT realized_pnl, realized_pnl_usdt, close_index_px,
|
||
created_at, closed_at, inst_id, underlying, premium_ccy
|
||
FROM options_trades
|
||
WHERE status = 'closed'
|
||
AND (realized_pnl IS NOT NULL OR realized_pnl_usdt IS NOT NULL)
|
||
"""
|
||
|
||
|
||
def freeze_missing_realized_pnl_usdt(conn, ex: Any = None) -> int:
|
||
"""旧已平仓缺冻结 U 时按 close_index/当前指数写一次,之后不再跟盘变.
|
||
|
||
注意:已有 realized_pnl 视为净额直接冻结,不再二次扣 open_fee/close_fee
|
||
(避免对 OKX realizedPnl 或已扣费本地净额重复减费).
|
||
"""
|
||
init_options_tables(conn)
|
||
rows = conn.execute(
|
||
"""
|
||
SELECT id, realized_pnl, realized_pnl_usdt, close_index_px,
|
||
inst_id, underlying, premium_ccy
|
||
FROM options_trades
|
||
WHERE status = 'closed'
|
||
AND realized_pnl_usdt IS NULL
|
||
AND realized_pnl IS NOT NULL
|
||
ORDER BY id DESC
|
||
LIMIT 300
|
||
"""
|
||
).fetchall()
|
||
if not rows:
|
||
return 0
|
||
from lib.options.options_pnl_settle_lib import (
|
||
freeze_pnl_to_usdt,
|
||
resolve_close_index_px,
|
||
)
|
||
|
||
updated = 0
|
||
for row in rows:
|
||
ccy = str(_row_get(row, "premium_ccy") or "").strip().upper()
|
||
idx = _safe_float(_row_get(row, "close_index_px"))
|
||
if idx is None or idx <= 0:
|
||
idx = resolve_close_index_px(
|
||
ex,
|
||
inst_id=str(_row_get(row, "inst_id") or ""),
|
||
underlying=str(_row_get(row, "underlying") or ""),
|
||
)
|
||
pnl = _safe_float(_row_get(row, "realized_pnl"))
|
||
if pnl is None:
|
||
continue
|
||
settled = freeze_pnl_to_usdt(pnl, premium_ccy=ccy or "USDC", index_px=idx)
|
||
usdt = settled.get("realized_pnl_usdt")
|
||
close_idx = settled.get("close_index_px")
|
||
if usdt is None:
|
||
continue
|
||
conn.execute(
|
||
"""
|
||
UPDATE options_trades
|
||
SET realized_pnl_usdt = ?,
|
||
close_index_px = COALESCE(close_index_px, ?)
|
||
WHERE id = ? AND realized_pnl_usdt IS NULL
|
||
""",
|
||
(float(usdt), close_idx, int(row["id"])),
|
||
)
|
||
updated += 1
|
||
return updated
|
||
|
||
|
||
def repair_double_counted_fee_pnl(conn, ex: Any = None) -> int:
|
||
"""修复「交易所净盈亏上又减了一次手续费」导致的偏亏冻结值.
|
||
|
||
判定:存在开/平手续费,且 realized_pnl ≈ (recv-paid) - 2×fees,则改回只扣一次.
|
||
若无可靠 recv/paid,但 usdt ≈ (realized + fees)×index 的反向不符则跳过.
|
||
"""
|
||
init_options_tables(conn)
|
||
rows = conn.execute(
|
||
"""
|
||
SELECT id, realized_pnl, realized_pnl_usdt, close_index_px,
|
||
inst_id, underlying, premium_ccy, premium_paid, premium_received,
|
||
open_fee, close_fee
|
||
FROM options_trades
|
||
WHERE status = 'closed'
|
||
AND (COALESCE(open_fee, 0) > 0 OR COALESCE(close_fee, 0) > 0)
|
||
AND premium_paid IS NOT NULL
|
||
AND premium_received IS NOT NULL
|
||
ORDER BY id DESC
|
||
LIMIT 300
|
||
"""
|
||
).fetchall()
|
||
if not rows:
|
||
return 0
|
||
from lib.options.options_pnl_settle_lib import resolve_close_index_px, settle_closed_pnl
|
||
|
||
updated = 0
|
||
for row in rows:
|
||
paid = _safe_float(_row_get(row, "premium_paid"))
|
||
recv = _safe_float(_row_get(row, "premium_received"))
|
||
ofee = abs(_safe_float(_row_get(row, "open_fee")) or 0.0)
|
||
cfee = abs(_safe_float(_row_get(row, "close_fee")) or 0.0)
|
||
fees = ofee + cfee
|
||
if paid is None or recv is None or fees <= 0:
|
||
continue
|
||
gross = float(recv) - float(paid)
|
||
single = round(gross - fees, 8)
|
||
double = round(gross - 2.0 * fees, 8)
|
||
cur = _safe_float(_row_get(row, "realized_pnl"))
|
||
usdt = _safe_float(_row_get(row, "realized_pnl_usdt"))
|
||
ccy = str(_row_get(row, "premium_ccy") or "").strip().upper() or "ETH"
|
||
idx = _safe_float(_row_get(row, "close_index_px"))
|
||
if idx is None or idx <= 0:
|
||
idx = resolve_close_index_px(
|
||
ex,
|
||
inst_id=str(_row_get(row, "inst_id") or ""),
|
||
underlying=str(_row_get(row, "underlying") or ""),
|
||
)
|
||
# 币数量层面像双重扣费,或 U 层面像双重扣费
|
||
coin_double = cur is not None and abs(cur - double) <= max(1e-8, abs(fees) * 1e-6)
|
||
usdt_double = False
|
||
if usdt is not None and idx is not None and idx > 0 and ccy in ("ETH", "BTC"):
|
||
usdt_double = abs(usdt - double * float(idx)) <= max(0.02, abs(fees * float(idx)) * 0.05)
|
||
if not coin_double and not usdt_double:
|
||
# 若当前已是单次扣费则不动
|
||
continue
|
||
settled = settle_closed_pnl(
|
||
premium_paid=paid,
|
||
premium_received=recv,
|
||
open_fee=ofee,
|
||
close_fee=cfee,
|
||
premium_ccy=ccy,
|
||
index_px=idx,
|
||
)
|
||
if settled.get("realized_pnl_usdt") is None and settled.get("realized_pnl") is None:
|
||
continue
|
||
conn.execute(
|
||
"""
|
||
UPDATE options_trades
|
||
SET realized_pnl = ?,
|
||
close_index_px = COALESCE(?, close_index_px),
|
||
realized_pnl_usdt = ?
|
||
WHERE id = ?
|
||
""",
|
||
(
|
||
settled.get("realized_pnl"),
|
||
settled.get("close_index_px"),
|
||
settled.get("realized_pnl_usdt"),
|
||
int(row["id"]),
|
||
),
|
||
)
|
||
updated += 1
|
||
return updated
|
||
|
||
|
||
def load_local_options_stats_history(conn, ex: Any = None) -> list[dict[str, Any]]:
|
||
"""本地已平(含冻结 U/手续费净额) + 当前 open,供统计/中控窗口复用."""
|
||
init_options_tables(conn)
|
||
try:
|
||
repair_double_counted_fee_pnl(conn, ex)
|
||
except Exception:
|
||
pass
|
||
try:
|
||
freeze_missing_realized_pnl_usdt(conn, ex)
|
||
except Exception:
|
||
pass
|
||
closed_rows = conn.execute(_CLOSED_STATS_SELECT).fetchall()
|
||
open_rows = conn.execute(
|
||
"SELECT created_at FROM options_trades WHERE status = 'open'"
|
||
).fetchall()
|
||
history = _closed_trade_dicts(closed_rows)
|
||
for row in open_rows:
|
||
history.append({"status": "open", "created_at": _row_get(row, "created_at")})
|
||
return history
|
||
|
||
|
||
def compute_options_stats(get_db, ex: Any = None) -> dict[str, Any]:
|
||
conn = get_db()
|
||
try:
|
||
history = load_local_options_stats_history(conn, ex)
|
||
try:
|
||
conn.commit()
|
||
except Exception:
|
||
pass
|
||
# 优先冻结 U;缺省再按指数折(与 /api/options/stats 一致)
|
||
return compute_options_stats_from_history(history_pnl_to_usdt(history, ex))
|
||
finally:
|
||
conn.close()
|
||
|
||
|
||
def _stats_from_option_trade_rows(closed_rows, open_rows, *, ex: Any = None) -> dict[str, Any]:
|
||
"""兼容旧调用;内部统一折 U 后走 compute_options_stats_from_history."""
|
||
history = _closed_trade_dicts(closed_rows)
|
||
for row in open_rows or []:
|
||
history.append({"status": "open", "created_at": _row_get(row, "created_at")})
|
||
return compute_options_stats_from_history(history_pnl_to_usdt(history, ex))
|
||
|
||
|
||
def header_options_stats_for_window(
|
||
conn,
|
||
list_window: dict[str, Any],
|
||
app_tz,
|
||
ex: Any = None,
|
||
) -> dict[str, Any]:
|
||
"""顶栏总交易/胜率/盈亏比:纯期权(options_trades),全站导航共用,按列表窗过滤.
|
||
|
||
币本位 realized_pnl 为币数量,须折 U 后再算盈亏比,否则 BTC/ETH 混算会严重失真.
|
||
"""
|
||
from lib.common.history_window_lib import utc_window_to_bj_sql_strings
|
||
|
||
init_options_tables(conn)
|
||
start_bj, end_bj = utc_window_to_bj_sql_strings(
|
||
list_window["start_utc"], list_window["end_utc"], app_tz
|
||
)
|
||
try:
|
||
freeze_missing_realized_pnl_usdt(conn, ex)
|
||
except Exception:
|
||
pass
|
||
closed_rows = conn.execute(
|
||
"""
|
||
SELECT realized_pnl, realized_pnl_usdt, close_index_px,
|
||
created_at, closed_at, inst_id, underlying, premium_ccy
|
||
FROM options_trades
|
||
WHERE status = 'closed'
|
||
AND (realized_pnl IS NOT NULL OR realized_pnl_usdt IS NOT NULL)
|
||
AND COALESCE(closed_at, created_at) >= ?
|
||
AND COALESCE(closed_at, created_at) <= ?
|
||
""",
|
||
(start_bj, end_bj),
|
||
).fetchall()
|
||
open_rows = conn.execute(
|
||
"SELECT created_at FROM options_trades WHERE status = 'open'"
|
||
).fetchall()
|
||
history = _closed_trade_dicts(closed_rows)
|
||
for row in open_rows:
|
||
history.append({"status": "open", "created_at": _row_get(row, "created_at")})
|
||
stats = compute_options_stats_from_history(history_pnl_to_usdt(history, ex))
|
||
return {
|
||
"total": int(stats.get("total_closed") or 0),
|
||
"rate": float(stats.get("win_rate") or 0),
|
||
"profit_loss_ratio": stats.get("profit_loss_ratio"),
|
||
}
|