dekun
f19500bcd9
Match exchange history by sheets and open time so an earlier close is not overwritten with the later trade's PnL. Co-authored-by: Cursor <cursoragent@cursor.com>
476 lines
16 KiB
Python
476 lines
16 KiB
Python
"""期权持仓监控:浮盈翻倍微信提醒 + 平仓/到期状态同步."""
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import sqlite3
|
|
import time
|
|
from datetime import datetime, timezone
|
|
from typing import Any, Callable
|
|
from zoneinfo import ZoneInfo
|
|
|
|
from lib.exchange.okx_options_lib import normalize_option_exp_ms, resolve_option_close_from_history
|
|
|
|
_APP_TZ = ZoneInfo((os.getenv("APP_TIMEZONE") or os.getenv("TZ") or "Asia/Shanghai").strip() or "Asia/Shanghai")
|
|
|
|
|
|
def _safe_float(v: Any) -> float | None:
|
|
if v is None:
|
|
return None
|
|
try:
|
|
return float(v)
|
|
except (TypeError, ValueError):
|
|
return None
|
|
|
|
|
|
def build_profit_alert_message(
|
|
*,
|
|
account_label: str,
|
|
inst_id: str,
|
|
premium_paid: float,
|
|
upl: float,
|
|
upl_ratio: float | None,
|
|
bid: float | None,
|
|
) -> str:
|
|
pct = f"{upl_ratio * 100:.1f}%" if upl_ratio is not None else "—"
|
|
bid_txt = f"{bid:.4f}" if bid is not None else "—"
|
|
return "\n".join(
|
|
[
|
|
"【OKX期权·翻倍提醒】",
|
|
f"账户:{account_label}",
|
|
f"合约:{inst_id}",
|
|
f"已付权利金:{premium_paid:.4f} USDC",
|
|
f"未实现盈亏:{upl:+.4f} USDC({pct})",
|
|
f"当前买一:{bid_txt}(可考虑限价平仓锁利)",
|
|
]
|
|
)
|
|
|
|
|
|
def run_options_profit_alerts(
|
|
conn: sqlite3.Connection,
|
|
positions: list[dict[str, Any]],
|
|
*,
|
|
profit_ratio: float,
|
|
send_wechat: Callable[[str], None],
|
|
account_label: str,
|
|
ticker_bid_fn: Callable[[str], float | None],
|
|
) -> int:
|
|
"""
|
|
对比 DB 中 open 记录与交易所持仓;达到阈值发微信.
|
|
返回发送条数.
|
|
"""
|
|
sent = 0
|
|
pos_by_inst = {str(p.get("inst_id") or p.get("instId") or ""): p for p in positions}
|
|
rows = conn.execute(
|
|
"""
|
|
SELECT id, inst_id, premium_paid, profit_alert_sent
|
|
FROM options_trades
|
|
WHERE status = 'open'
|
|
ORDER BY id ASC
|
|
"""
|
|
).fetchall()
|
|
# 同合约多腿加仓:按合约汇总权利金,整仓只告警一次
|
|
by_inst: dict[str, dict[str, Any]] = {}
|
|
for row in rows:
|
|
inst_id = str(row["inst_id"] or "")
|
|
if not inst_id:
|
|
continue
|
|
bucket = by_inst.setdefault(
|
|
inst_id,
|
|
{"ids": [], "premium": 0.0, "all_sent": True, "has_prem": False},
|
|
)
|
|
bucket["ids"].append(int(row["id"]))
|
|
prem = _safe_float(row["premium_paid"])
|
|
if prem is not None:
|
|
bucket["premium"] += float(prem)
|
|
bucket["has_prem"] = True
|
|
if not int(row["profit_alert_sent"] or 0):
|
|
bucket["all_sent"] = False
|
|
|
|
for inst_id, bucket in by_inst.items():
|
|
if bucket["all_sent"] or not bucket["has_prem"] or bucket["premium"] <= 0:
|
|
continue
|
|
pos = pos_by_inst.get(inst_id)
|
|
if not pos:
|
|
continue
|
|
prem = float(bucket["premium"])
|
|
upl = _safe_float(pos.get("upl"))
|
|
upl_ratio = _safe_float(pos.get("upl_ratio_pct"))
|
|
if upl_ratio is not None:
|
|
ratio = upl_ratio / 100.0
|
|
elif upl is not None:
|
|
ratio = upl / prem
|
|
else:
|
|
continue
|
|
if ratio < float(profit_ratio):
|
|
continue
|
|
bid = ticker_bid_fn(inst_id)
|
|
msg = build_profit_alert_message(
|
|
account_label=account_label,
|
|
inst_id=inst_id,
|
|
premium_paid=prem,
|
|
upl=upl or 0.0,
|
|
upl_ratio=ratio,
|
|
bid=bid,
|
|
)
|
|
try:
|
|
send_wechat(msg)
|
|
conn.execute(
|
|
f"UPDATE options_trades SET profit_alert_sent = 1 WHERE id IN ({','.join('?' * len(bucket['ids']))})",
|
|
tuple(bucket["ids"]),
|
|
)
|
|
sent += 1
|
|
except Exception:
|
|
pass
|
|
return sent
|
|
|
|
|
|
def _created_at_ms(created_at: Any) -> int | None:
|
|
"""墙钟 created_at → UTC ms.库内时间为业务时区(默认 Asia/Shanghai),不可当 UTC."""
|
|
if not created_at:
|
|
return None
|
|
raw = str(created_at).strip()
|
|
if not raw:
|
|
return None
|
|
for fmt, ln in (("%Y-%m-%d %H:%M:%S", 19), ("%Y-%m-%d %H:%M:%f", 26), ("%Y-%m-%d %H:%M", 16)):
|
|
try:
|
|
dt = datetime.strptime(raw[:ln], fmt).replace(tzinfo=_APP_TZ)
|
|
return int(dt.timestamp() * 1000)
|
|
except ValueError:
|
|
continue
|
|
return None
|
|
|
|
|
|
def _group_key_for_closed_trade(row: Any) -> str:
|
|
inst = str(row["inst_id"] or "").strip()
|
|
closed = str(row["closed_at"] or "").strip()
|
|
close_prefix = closed[:16] if closed else ""
|
|
ord_id = str(row["close_ord_id"] or "").strip() if "close_ord_id" in row.keys() else ""
|
|
# 即使 close_ord_id/posId 相同,也要按平仓时间拆开(OKX 可能复用 posId)
|
|
if ord_id:
|
|
return f"{inst}|ord:{ord_id}|close:{close_prefix}"
|
|
return f"{inst}|close:{close_prefix}"
|
|
|
|
|
|
def backfill_closed_options_realized_pnl_from_history(
|
|
conn: sqlite3.Connection,
|
|
hist_rows: list[dict[str, Any]],
|
|
*,
|
|
trade_limit: int = 200,
|
|
) -> int:
|
|
"""
|
|
用 OKX positions-history 的 realizedPnl 覆盖本地已平记录.
|
|
同一次平仓多笔本地 open(加仓)按权利金占比分摊交易所总盈亏.
|
|
"""
|
|
by_inst: dict[str, list[dict[str, Any]]] = {}
|
|
for raw in hist_rows or []:
|
|
if not isinstance(raw, dict):
|
|
continue
|
|
inst = str(raw.get("instId") or "").strip()
|
|
if not inst:
|
|
continue
|
|
by_inst.setdefault(inst, []).append(raw)
|
|
|
|
rows = conn.execute(
|
|
"""
|
|
SELECT id, inst_id, sheets, premium_paid, realized_pnl, close_quote,
|
|
created_at, closed_at, close_ord_id
|
|
FROM options_trades
|
|
WHERE status = 'closed'
|
|
ORDER BY id DESC
|
|
LIMIT ?
|
|
""",
|
|
(int(trade_limit),),
|
|
).fetchall()
|
|
if not rows:
|
|
return 0
|
|
|
|
groups: dict[str, list[Any]] = {}
|
|
for row in rows:
|
|
inst = str(row["inst_id"] or "").strip()
|
|
if not inst or inst not in by_inst:
|
|
continue
|
|
groups.setdefault(_group_key_for_closed_trade(row), []).append(row)
|
|
|
|
updated = 0
|
|
for group in groups.values():
|
|
inst = str(group[0]["inst_id"] or "").strip()
|
|
open_candidates = [_created_at_ms(r["created_at"]) for r in group]
|
|
open_ms = min((x for x in open_candidates if x is not None), default=None)
|
|
close_candidates = [_created_at_ms(r["closed_at"]) for r in group]
|
|
close_ms = max((x for x in close_candidates if x is not None), default=None)
|
|
sheets_hint = None
|
|
try:
|
|
sheets_hint = sum(float(_safe_float(r["sheets"]) or 0.0) for r in group) or None
|
|
except (TypeError, ValueError):
|
|
sheets_hint = None
|
|
close_info = resolve_option_close_from_history(
|
|
by_inst.get(inst) or [],
|
|
open_ms=open_ms,
|
|
close_ms=close_ms,
|
|
sheets=sheets_hint,
|
|
)
|
|
if not close_info:
|
|
continue
|
|
ex_pnl = _safe_float(close_info.get("realized_pnl"))
|
|
if ex_pnl is None:
|
|
continue
|
|
close_quote = _safe_float(close_info.get("close_quote"))
|
|
matched_pos = str(close_info.get("pos_id") or "").strip() or None
|
|
total_paid = 0.0
|
|
for r in group:
|
|
total_paid += float(_safe_float(r["premium_paid"]) or 0.0)
|
|
allocated = 0.0
|
|
for i, r in enumerate(group):
|
|
paid = float(_safe_float(r["premium_paid"]) or 0.0)
|
|
if i == len(group) - 1:
|
|
share = round(float(ex_pnl) - allocated, 4)
|
|
elif total_paid > 0:
|
|
share = round(float(ex_pnl) * (paid / total_paid), 4)
|
|
allocated += share
|
|
else:
|
|
share = round(float(ex_pnl) / len(group), 4)
|
|
allocated += share
|
|
local = _safe_float(r["realized_pnl"])
|
|
local_close = _safe_float(r["close_quote"])
|
|
local_ord = str(r["close_ord_id"] or "").strip()
|
|
pnl_ok = local is not None and abs(local - share) < 1e-6
|
|
quote_ok = close_quote is None or (
|
|
local_close is not None and abs(local_close - float(close_quote)) < 1e-6
|
|
)
|
|
ord_ok = (not matched_pos) or (local_ord == matched_pos)
|
|
if pnl_ok and quote_ok and ord_ok:
|
|
continue
|
|
prem_recv = round(paid + share, 4)
|
|
conn.execute(
|
|
"""
|
|
UPDATE options_trades
|
|
SET realized_pnl = ?,
|
|
premium_received = ?,
|
|
close_quote = COALESCE(?, close_quote),
|
|
close_ord_id = COALESCE(?, close_ord_id)
|
|
WHERE id = ?
|
|
""",
|
|
(share, prem_recv, close_quote, matched_pos, int(r["id"])),
|
|
)
|
|
updated += 1
|
|
return updated
|
|
|
|
|
|
def sync_open_options_trades(
|
|
conn: sqlite3.Connection,
|
|
*,
|
|
live_inst_ids: set[str],
|
|
fetch_history_fn: Callable[[str], list[dict[str, Any]]],
|
|
notify_cfg: dict[str, Any] | None = None,
|
|
) -> int:
|
|
"""
|
|
交易所已无持仓时,将本地 open 记录同步为 closed.
|
|
优先用 positions-history 回填盈亏;否则到期后按归零处理.
|
|
"""
|
|
rows = conn.execute(
|
|
"""
|
|
SELECT id, inst_id, premium_paid, exp_time, created_at
|
|
FROM options_trades
|
|
WHERE status = 'open'
|
|
"""
|
|
).fetchall()
|
|
updated = 0
|
|
now_ms = int(time.time() * 1000)
|
|
for row in rows:
|
|
inst_id = str(row["inst_id"] or "")
|
|
if not inst_id or inst_id in live_inst_ids:
|
|
continue
|
|
paid = _safe_float(row["premium_paid"]) or 0.0
|
|
open_ms = _created_at_ms(row["created_at"])
|
|
exp_ms = normalize_option_exp_ms(row["exp_time"], inst_id)
|
|
close_quote: float | None = None
|
|
prem_recv: float | None = None
|
|
realized_pnl: float | None = None
|
|
close_ord_id: str | None = None
|
|
closed_at: str | None = None
|
|
close_reason = "exchange"
|
|
|
|
close_info = resolve_option_close_from_history(
|
|
fetch_history_fn(inst_id),
|
|
open_ms=open_ms,
|
|
)
|
|
if close_info:
|
|
close_quote = close_info.get("close_quote")
|
|
realized_pnl = close_info.get("realized_pnl")
|
|
close_ord_id = close_info.get("pos_id")
|
|
if realized_pnl is not None:
|
|
prem_recv = round(paid + float(realized_pnl), 4)
|
|
close_ms = close_info.get("close_ms")
|
|
if close_ms:
|
|
closed_at = datetime.fromtimestamp(int(close_ms) / 1000, tz=timezone.utc).strftime(
|
|
"%Y-%m-%d %H:%M:%S"
|
|
)
|
|
elif exp_ms is not None and now_ms >= int(exp_ms):
|
|
close_reason = "expired"
|
|
close_quote = 0.0
|
|
prem_recv = 0.0
|
|
realized_pnl = round(-paid, 4)
|
|
if exp_ms:
|
|
closed_at = datetime.fromtimestamp(int(exp_ms) / 1000, tz=timezone.utc).strftime(
|
|
"%Y-%m-%d %H:%M:%S"
|
|
)
|
|
else:
|
|
continue
|
|
|
|
conn.execute(
|
|
"""
|
|
UPDATE options_trades
|
|
SET status = 'closed',
|
|
close_quote = ?,
|
|
premium_received = ?,
|
|
realized_pnl = ?,
|
|
close_ord_id = COALESCE(?, close_ord_id),
|
|
closed_at = COALESCE(?, closed_at, CURRENT_TIMESTAMP),
|
|
signal_note = CASE
|
|
WHEN ? = 'expired' AND (signal_note IS NULL OR TRIM(signal_note) = '')
|
|
THEN '到期结算'
|
|
ELSE signal_note
|
|
END
|
|
WHERE id = ?
|
|
""",
|
|
(
|
|
close_quote,
|
|
prem_recv,
|
|
realized_pnl,
|
|
close_ord_id,
|
|
closed_at,
|
|
close_reason,
|
|
int(row["id"]),
|
|
),
|
|
)
|
|
updated += 1
|
|
if notify_cfg is not None:
|
|
try:
|
|
from lib.options.options_notify_lib import notify_options_close
|
|
|
|
reason = "到期结算" if close_reason == "expired" else "交易所平仓"
|
|
notify_options_close(
|
|
notify_cfg,
|
|
conn,
|
|
inst_id=inst_id,
|
|
reason=reason,
|
|
trade_id=int(row["id"]),
|
|
premium_paid=paid,
|
|
premium_received=prem_recv,
|
|
realized_pnl=realized_pnl,
|
|
close_quote=close_quote,
|
|
)
|
|
except Exception:
|
|
pass
|
|
return updated
|
|
|
|
|
|
def reconcile_live_open_trades(
|
|
conn: sqlite3.Connection,
|
|
*,
|
|
live_inst_ids: set[str],
|
|
) -> int:
|
|
"""交易所有持仓但本地误标 closed 时恢复为 open."""
|
|
fixed = 0
|
|
for inst_id in live_inst_ids:
|
|
if not inst_id:
|
|
continue
|
|
open_row = conn.execute(
|
|
"SELECT id FROM options_trades WHERE inst_id = ? AND status = 'open' LIMIT 1",
|
|
(inst_id,),
|
|
).fetchone()
|
|
if open_row:
|
|
continue
|
|
row = conn.execute(
|
|
"""
|
|
SELECT id, close_ord_id, realized_pnl
|
|
FROM options_trades
|
|
WHERE inst_id = ? AND status = 'closed'
|
|
ORDER BY id DESC LIMIT 1
|
|
""",
|
|
(inst_id,),
|
|
).fetchone()
|
|
if not row:
|
|
continue
|
|
if row["close_ord_id"]:
|
|
continue
|
|
if row["realized_pnl"] is not None:
|
|
continue
|
|
conn.execute(
|
|
"""
|
|
UPDATE options_trades
|
|
SET status = 'open',
|
|
close_quote = NULL,
|
|
premium_received = NULL,
|
|
realized_pnl = NULL,
|
|
closed_at = NULL,
|
|
signal_note = CASE
|
|
WHEN signal_note = '到期结算' THEN NULL
|
|
ELSE signal_note
|
|
END
|
|
WHERE id = ?
|
|
""",
|
|
(int(row["id"]),),
|
|
)
|
|
fixed += 1
|
|
return fixed
|
|
|
|
|
|
def options_monitor_loop(
|
|
*,
|
|
enabled: bool,
|
|
poll_seconds: float,
|
|
get_db: Callable[[], sqlite3.Connection],
|
|
fetch_positions: Callable[[], list[dict[str, Any]]],
|
|
ticker_bid_fn: Callable[[str], float | None],
|
|
send_wechat: Callable[[str], None],
|
|
account_label: str,
|
|
profit_ratio: float,
|
|
sync_trades_fn: Callable[[sqlite3.Connection], int] | None = None,
|
|
target_close_fn: Callable[[str], dict[str, Any]] | None = None,
|
|
stale_pending_fn: Callable[[], dict[str, Any]] | None = None,
|
|
stop_event: Any = None,
|
|
) -> None:
|
|
if not enabled:
|
|
return
|
|
while True:
|
|
if stop_event is not None and getattr(stop_event, "is_set", lambda: False)():
|
|
break
|
|
try:
|
|
conn = get_db()
|
|
try:
|
|
positions = fetch_positions()
|
|
run_options_profit_alerts(
|
|
conn,
|
|
positions,
|
|
profit_ratio=profit_ratio,
|
|
send_wechat=send_wechat,
|
|
account_label=account_label,
|
|
ticker_bid_fn=ticker_bid_fn,
|
|
)
|
|
if target_close_fn is not None:
|
|
from lib.options.options_target_lib import run_options_target_closes
|
|
|
|
run_options_target_closes(
|
|
conn,
|
|
positions,
|
|
close_fn=target_close_fn,
|
|
send_wechat=send_wechat,
|
|
account_label=account_label,
|
|
cfg={"send_wechat": send_wechat, "account_label": account_label},
|
|
)
|
|
if sync_trades_fn is not None:
|
|
sync_trades_fn(conn)
|
|
conn.commit()
|
|
finally:
|
|
conn.close()
|
|
# 平仓限价挂单超时撤单(独立于 DB 事务)
|
|
if stale_pending_fn is not None:
|
|
try:
|
|
stale_pending_fn()
|
|
except Exception:
|
|
pass
|
|
except Exception:
|
|
pass
|
|
time.sleep(max(5.0, float(poll_seconds)))
|