dekun
12574ae88a
Treat the filter as fuzzy q over underlying/inst/strategy (BTCUSDT->BTC) instead of exact strategy_tag, which always wiped pending rows. Co-authored-by: Cursor <cursoragent@cursor.com>
1010 lines
35 KiB
Python
1010 lines
35 KiB
Python
"""期权复盘业务:OKX 已平期权导入 + 已结束对冲计划导入 + 复盘 CRUD + 统计."""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import sqlite3
|
|
from datetime import datetime
|
|
from typing import Any, Callable, Optional
|
|
|
|
from lib.options.options_review_db import (
|
|
SOURCE_OPTION,
|
|
SOURCE_OPTIONS_OPTIONS,
|
|
SOURCE_PERP_OPTIONS,
|
|
SOURCE_TYPES,
|
|
init_options_review_tables,
|
|
)
|
|
from lib.options.options_review_images_lib import (
|
|
images_json_dumps,
|
|
parse_options_review_images_json,
|
|
)
|
|
|
|
SOURCE_LABELS = {
|
|
SOURCE_OPTION: "纯期权",
|
|
SOURCE_PERP_OPTIONS: "永期对冲",
|
|
SOURCE_OPTIONS_OPTIONS: "期期对冲",
|
|
}
|
|
|
|
HOLD_BUCKETS = (
|
|
("0-1h", 0, 3600),
|
|
("1-6h", 3600, 6 * 3600),
|
|
("6-24h", 6 * 3600, 24 * 3600),
|
|
("1-3d", 24 * 3600, 3 * 24 * 3600),
|
|
(">3d", 3 * 24 * 3600, None),
|
|
)
|
|
|
|
|
|
def _now_str() -> str:
|
|
return datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
|
|
|
|
|
def _parse_ts(raw: Any) -> Optional[datetime]:
|
|
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(opened_at: Any, closed_at: Any) -> Optional[int]:
|
|
start = _parse_ts(opened_at)
|
|
end = _parse_ts(closed_at)
|
|
if start is None or end is None:
|
|
return None
|
|
sec = int((end - start).total_seconds())
|
|
return sec if sec >= 0 else None
|
|
|
|
|
|
def _safe_float(v: Any) -> Optional[float]:
|
|
if v is None or v == "":
|
|
return None
|
|
try:
|
|
return float(v)
|
|
except (TypeError, ValueError):
|
|
return None
|
|
|
|
|
|
def get_sync_state(conn: sqlite3.Connection, key: str) -> Optional[str]:
|
|
row = conn.execute(
|
|
"SELECT value FROM options_review_sync_state WHERE key=?", (key,)
|
|
).fetchone()
|
|
return str(row["value"]) if row and row["value"] is not None else None
|
|
|
|
|
|
def set_sync_state(conn: sqlite3.Connection, key: str, value: str) -> None:
|
|
conn.execute(
|
|
"""
|
|
INSERT INTO options_review_sync_state(key, value, updated_at)
|
|
VALUES (?, ?, ?)
|
|
ON CONFLICT(key) DO UPDATE SET value=excluded.value, updated_at=excluded.updated_at
|
|
""",
|
|
(key, value, _now_str()),
|
|
)
|
|
|
|
|
|
def _purge_review_trade_by_key(conn: sqlite3.Connection, history_key: str) -> bool:
|
|
"""删除已导入的复盘快照(含复盘内容)."""
|
|
key = str(history_key or "").strip()
|
|
if not key:
|
|
return False
|
|
existing = conn.execute(
|
|
"SELECT id FROM options_review_trades WHERE history_key=?", (key,)
|
|
).fetchone()
|
|
if not existing:
|
|
return False
|
|
tid = int(existing["id"])
|
|
conn.execute("DELETE FROM options_review_entries WHERE trade_id=?", (tid,))
|
|
conn.execute("DELETE FROM options_review_trades WHERE id=?", (tid,))
|
|
return True
|
|
|
|
|
|
def upsert_option_history_row(conn: sqlite3.Connection, row: dict[str, Any]) -> str:
|
|
"""幂等写入纯期权快照;不触碰 options_review_entries;已隐藏的不再导入."""
|
|
history_key = str(row.get("history_key") or "").strip()
|
|
if not history_key:
|
|
return "skip"
|
|
if is_review_hidden(
|
|
conn,
|
|
history_key,
|
|
inst_id=str(row.get("inst_id") or "").strip() or None,
|
|
closed_at=row.get("closed_at") or row.get("created_at"),
|
|
):
|
|
# 若此前已导入,清掉,避免列表残留
|
|
return "purged" if _purge_review_trade_by_key(conn, history_key) else "hidden"
|
|
opened_at = row.get("created_at") or row.get("opened_at")
|
|
closed_at = row.get("closed_at")
|
|
pnl = _safe_float(row.get("realized_pnl"))
|
|
hold = _hold_seconds(opened_at, closed_at)
|
|
existing = conn.execute(
|
|
"SELECT id FROM options_review_trades WHERE history_key=?", (history_key,)
|
|
).fetchone()
|
|
fields = {
|
|
"source_type": SOURCE_OPTION,
|
|
"history_key": history_key,
|
|
"underlying": str(row.get("underlying") or "").strip() or None,
|
|
"opened_at": opened_at,
|
|
"closed_at": closed_at,
|
|
"hold_seconds": hold,
|
|
"realized_pnl_total": pnl,
|
|
"status_raw": str(row.get("status_label") or row.get("status") or "closed"),
|
|
"synced_at": _now_str(),
|
|
"pos_id": str(row.get("pos_id") or "").strip() or None,
|
|
"inst_id": str(row.get("inst_id") or "").strip() or None,
|
|
"opt_type": str(row.get("opt_type") or "").strip() or None,
|
|
"strike": _safe_float(row.get("strike")),
|
|
"exp_time": str(row.get("exp_time") or "").strip() or None,
|
|
"sheets": int(row.get("sheets") or 0) or None,
|
|
"open_avg": _safe_float(row.get("open_avg_px") if row.get("open_avg_px") is not None else row.get("open_avg")),
|
|
"close_avg": _safe_float(row.get("close_avg_px") if row.get("close_avg_px") is not None else row.get("close_avg")),
|
|
"premium_paid": _safe_float(row.get("premium_paid")),
|
|
"realized_pnl": pnl,
|
|
}
|
|
cols = list(fields.keys())
|
|
if existing:
|
|
sets = ", ".join(f"{c}=?" for c in cols if c != "history_key")
|
|
vals = [fields[c] for c in cols if c != "history_key"]
|
|
conn.execute(
|
|
f"UPDATE options_review_trades SET {sets} WHERE history_key=?",
|
|
[*vals, history_key],
|
|
)
|
|
return "updated"
|
|
placeholders = ",".join(["?"] * len(cols))
|
|
conn.execute(
|
|
f"INSERT INTO options_review_trades ({','.join(cols)}) VALUES ({placeholders})",
|
|
[fields[c] for c in cols],
|
|
)
|
|
return "inserted"
|
|
|
|
|
|
def _close_fingerprint(inst_id: Any, closed_at: Any) -> str | None:
|
|
inst = str(inst_id or "").strip()
|
|
if not inst:
|
|
return None
|
|
closed = str(closed_at or "").strip()
|
|
if not closed:
|
|
return f"inst:{inst}"
|
|
# 精确到分钟,避免秒差导致漏匹配
|
|
return f"inst_close:{inst}:{closed[:16]}"
|
|
|
|
|
|
def is_review_hidden(
|
|
conn: sqlite3.Connection,
|
|
history_key: str,
|
|
*,
|
|
inst_id: str | None = None,
|
|
closed_at: Any = None,
|
|
) -> bool:
|
|
init_options_review_tables(conn)
|
|
key = str(history_key or "").strip()
|
|
if key and conn.execute(
|
|
"SELECT 1 FROM options_review_hidden WHERE history_key=? LIMIT 1", (key,)
|
|
).fetchone():
|
|
return True
|
|
fp = _close_fingerprint(inst_id, closed_at)
|
|
if fp and conn.execute(
|
|
"SELECT 1 FROM options_review_hidden WHERE history_key=? LIMIT 1", (fp,)
|
|
).fetchone():
|
|
return True
|
|
# 期权历史页删除:options_history_hidden,按合约指纹或原 key
|
|
try:
|
|
if key and conn.execute(
|
|
"SELECT 1 FROM options_history_hidden WHERE history_key=? LIMIT 1", (key,)
|
|
).fetchone():
|
|
return True
|
|
if fp and conn.execute(
|
|
"SELECT 1 FROM options_history_hidden WHERE history_key=? LIMIT 1", (fp,)
|
|
).fetchone():
|
|
return True
|
|
# 仅隐藏了 ex:posId 时,用合约+平仓时间在历史隐藏表无直接命中;
|
|
# 若指纹已写入 options_review_hidden(新删除路径)上面已覆盖.
|
|
# 兼容:inst 级隐藏
|
|
if inst_id:
|
|
inst_fp = f"inst:{str(inst_id).strip()}"
|
|
if conn.execute(
|
|
"SELECT 1 FROM options_review_hidden WHERE history_key=? LIMIT 1",
|
|
(inst_fp,),
|
|
).fetchone():
|
|
return True
|
|
if conn.execute(
|
|
"SELECT 1 FROM options_history_hidden WHERE history_key=? LIMIT 1",
|
|
(inst_fp,),
|
|
).fetchone():
|
|
return True
|
|
except Exception:
|
|
pass
|
|
return False
|
|
|
|
|
|
def hide_review_keys(
|
|
conn: sqlite3.Connection,
|
|
*,
|
|
history_key: str,
|
|
inst_id: str | None = None,
|
|
closed_at: Any = None,
|
|
) -> None:
|
|
init_options_review_tables(conn)
|
|
keys = [str(history_key or "").strip()]
|
|
fp = _close_fingerprint(inst_id, closed_at)
|
|
if fp:
|
|
keys.append(fp)
|
|
for k in keys:
|
|
if not k:
|
|
continue
|
|
conn.execute(
|
|
"""
|
|
INSERT OR IGNORE INTO options_review_hidden(history_key, inst_id, closed_at)
|
|
VALUES (?, ?, ?)
|
|
""",
|
|
(k, (inst_id or None), str(closed_at or "")[:19] or None),
|
|
)
|
|
try:
|
|
conn.execute(
|
|
"INSERT OR IGNORE INTO options_history_hidden(history_key) VALUES (?)",
|
|
(k,),
|
|
)
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
def hide_review_trade(conn: sqlite3.Connection, trade_id: int) -> dict[str, Any]:
|
|
"""从复盘列表删除并持久隐藏,刷新本地源也不会再回来."""
|
|
init_options_review_tables(conn)
|
|
row = conn.execute(
|
|
"SELECT * FROM options_review_trades WHERE id=?", (int(trade_id),)
|
|
).fetchone()
|
|
if not row:
|
|
return {"ok": False, "msg": "记录不存在"}
|
|
d = _row_to_dict(row)
|
|
hide_review_keys(
|
|
conn,
|
|
history_key=str(d.get("history_key") or ""),
|
|
inst_id=str(d.get("inst_id") or "").strip() or None,
|
|
closed_at=d.get("closed_at") or d.get("opened_at"),
|
|
)
|
|
entry = conn.execute(
|
|
"SELECT * FROM options_review_entries WHERE trade_id=?", (int(trade_id),)
|
|
).fetchone()
|
|
conn.execute("DELETE FROM options_review_entries WHERE trade_id=?", (int(trade_id),))
|
|
conn.execute("DELETE FROM options_review_trades WHERE id=?", (int(trade_id),))
|
|
return {"ok": True, "entry": _row_to_dict(entry) if entry else None, "history_key": d.get("history_key")}
|
|
|
|
|
|
|
|
def sync_options_from_local_trades(conn: sqlite3.Connection) -> dict[str, Any]:
|
|
"""从本地 options_trades 已平仓记录导入复盘快照(不访问交易所)."""
|
|
init_options_review_tables(conn)
|
|
from lib.options.options_db import init_options_tables
|
|
|
|
init_options_tables(conn)
|
|
rows = conn.execute(
|
|
"""
|
|
SELECT id, inst_id, underlying, opt_type, strike, exp_time, sheets,
|
|
open_quote, close_quote, premium_paid, realized_pnl,
|
|
created_at, closed_at, signal_note, status
|
|
FROM options_trades
|
|
WHERE status = 'closed'
|
|
ORDER BY id DESC
|
|
LIMIT 500
|
|
"""
|
|
).fetchall()
|
|
inserted = updated = skipped = 0
|
|
for r in rows:
|
|
trade_id = int(r["id"])
|
|
history_key = f"local_opt:{trade_id}"
|
|
pnl = _safe_float(r["realized_pnl"])
|
|
opened_at = r["created_at"]
|
|
closed_at = r["closed_at"]
|
|
action = upsert_option_history_row(
|
|
conn,
|
|
{
|
|
"history_key": history_key,
|
|
"pos_id": f"local:{trade_id}",
|
|
"inst_id": r["inst_id"],
|
|
"underlying": r["underlying"],
|
|
"opt_type": r["opt_type"],
|
|
"strike": r["strike"],
|
|
"exp_time": r["exp_time"],
|
|
"sheets": r["sheets"],
|
|
"open_avg_px": r["open_quote"],
|
|
"close_avg_px": r["close_quote"],
|
|
"premium_paid": r["premium_paid"],
|
|
"realized_pnl": pnl,
|
|
"created_at": opened_at,
|
|
"closed_at": closed_at,
|
|
"status_label": "已平",
|
|
},
|
|
)
|
|
if action == "inserted":
|
|
inserted += 1
|
|
elif action == "updated":
|
|
updated += 1
|
|
else:
|
|
skipped += 1
|
|
set_sync_state(conn, "options_last_sync_at", _now_str())
|
|
set_sync_state(conn, "options_last_count", str(len(rows)))
|
|
set_sync_state(conn, "options_sync_source", "local")
|
|
return {
|
|
"ok": True,
|
|
"source": "local",
|
|
"fetched": len(rows),
|
|
"inserted": inserted,
|
|
"updated": updated,
|
|
"skipped": skipped,
|
|
}
|
|
|
|
|
|
def sync_options_from_exchange(
|
|
conn: sqlite3.Connection,
|
|
ex: Any,
|
|
*,
|
|
limit: int = 500,
|
|
fetch_fn: Optional[Callable[..., list[dict[str, Any]]]] = None,
|
|
format_fn: Optional[Callable[..., dict[str, Any]]] = None,
|
|
) -> dict[str, Any]:
|
|
"""从 OKX positions-history 导入已全平期权仓位(可选,默认不用)."""
|
|
init_options_review_tables(conn)
|
|
from lib.exchange.okx_options_lib import (
|
|
fetch_all_option_positions_history,
|
|
format_option_history_row,
|
|
tick_sz_and_ct_mult,
|
|
)
|
|
|
|
fetch = fetch_fn or fetch_all_option_positions_history
|
|
fmt = format_fn or format_option_history_row
|
|
raw_rows = fetch(ex, limit=limit)
|
|
meta_cache: dict[str, dict[str, Any] | None] = {}
|
|
inserted = updated = skipped = 0
|
|
for raw in raw_rows:
|
|
inst_id = str(raw.get("instId") or "").strip()
|
|
tick_sz, ct_mult = None, 0.01
|
|
try:
|
|
tick_sz, ct_mult = tick_sz_and_ct_mult(ex, inst_id, meta_cache)
|
|
except Exception:
|
|
pass
|
|
formatted = fmt(raw, tick_sz=tick_sz, ct_mult=ct_mult)
|
|
action = upsert_option_history_row(conn, formatted)
|
|
if action == "inserted":
|
|
inserted += 1
|
|
elif action == "updated":
|
|
updated += 1
|
|
else:
|
|
skipped += 1
|
|
set_sync_state(conn, "options_last_sync_at", _now_str())
|
|
set_sync_state(conn, "options_last_count", str(len(raw_rows)))
|
|
set_sync_state(conn, "options_sync_source", "exchange")
|
|
return {
|
|
"ok": True,
|
|
"source": "exchange",
|
|
"fetched": len(raw_rows),
|
|
"inserted": inserted,
|
|
"updated": updated,
|
|
"skipped": skipped,
|
|
}
|
|
|
|
|
|
def _legs_json_from_plan(legs: list[dict[str, Any]]) -> str:
|
|
slim = []
|
|
for leg in legs:
|
|
slim.append(
|
|
{
|
|
"id": leg.get("id"),
|
|
"leg_role": leg.get("leg_role"),
|
|
"symbol": leg.get("symbol"),
|
|
"inst_id": leg.get("inst_id"),
|
|
"opt_type": leg.get("opt_type"),
|
|
"strike": leg.get("strike"),
|
|
"side": leg.get("side"),
|
|
"size": leg.get("size"),
|
|
"avg_open": leg.get("avg_open"),
|
|
"premium": leg.get("premium"),
|
|
"status": leg.get("status"),
|
|
"realized_pnl": leg.get("realized_pnl"),
|
|
"close_reason": leg.get("close_reason"),
|
|
"opened_at": leg.get("opened_at"),
|
|
"closed_at": leg.get("closed_at"),
|
|
}
|
|
)
|
|
return json.dumps(slim, ensure_ascii=False, separators=(",", ":"))
|
|
|
|
|
|
def upsert_hedge_plan_row(
|
|
conn: sqlite3.Connection,
|
|
plan: dict[str, Any],
|
|
legs: list[dict[str, Any]],
|
|
) -> str:
|
|
plan_id = int(plan["id"])
|
|
history_key = f"hedge:{plan_id}"
|
|
plan_type = str(plan.get("plan_type") or "").strip()
|
|
if plan_type not in (SOURCE_PERP_OPTIONS, SOURCE_OPTIONS_OPTIONS):
|
|
return "skip"
|
|
opened_at = plan.get("opened_at") or plan.get("created_at")
|
|
closed_at = plan.get("closed_at")
|
|
if is_review_hidden(
|
|
conn,
|
|
history_key,
|
|
inst_id=None,
|
|
closed_at=closed_at,
|
|
):
|
|
return "purged" if _purge_review_trade_by_key(conn, history_key) else "hidden"
|
|
total = _safe_float(plan.get("realized_pnl_total"))
|
|
hold = _hold_seconds(opened_at, closed_at)
|
|
fields = {
|
|
"source_type": plan_type,
|
|
"history_key": history_key,
|
|
"underlying": str(plan.get("underlying") or "").strip() or None,
|
|
"opened_at": opened_at,
|
|
"closed_at": closed_at,
|
|
"hold_seconds": hold,
|
|
"realized_pnl_total": total,
|
|
"status_raw": str(plan.get("status") or "closed"),
|
|
"synced_at": _now_str(),
|
|
"hedge_plan_id": plan_id,
|
|
"plan_close_reason": str(plan.get("close_reason") or "").strip() or None,
|
|
"realized_pnl_perp": _safe_float(plan.get("realized_pnl_perp")),
|
|
"realized_pnl_options": _safe_float(plan.get("realized_pnl_options")),
|
|
"premium_total": _safe_float(plan.get("premium_total")),
|
|
"direction": str(plan.get("direction") or "").strip() or None,
|
|
"tp": _safe_float(plan.get("tp")),
|
|
"sl": _safe_float(plan.get("sl")),
|
|
"target_price": _safe_float(plan.get("target_price")),
|
|
"target_price_up": _safe_float(plan.get("target_price_up")),
|
|
"target_price_down": _safe_float(plan.get("target_price_down")),
|
|
"legs_json": _legs_json_from_plan(legs),
|
|
}
|
|
existing = conn.execute(
|
|
"SELECT id FROM options_review_trades WHERE history_key=?", (history_key,)
|
|
).fetchone()
|
|
cols = list(fields.keys())
|
|
if existing:
|
|
sets = ", ".join(f"{c}=?" for c in cols if c != "history_key")
|
|
vals = [fields[c] for c in cols if c != "history_key"]
|
|
conn.execute(
|
|
f"UPDATE options_review_trades SET {sets} WHERE history_key=?",
|
|
[*vals, history_key],
|
|
)
|
|
trade_id = int(existing["id"])
|
|
action = "updated"
|
|
else:
|
|
placeholders = ",".join(["?"] * len(cols))
|
|
cur = conn.execute(
|
|
f"INSERT INTO options_review_trades ({','.join(cols)}) VALUES ({placeholders})",
|
|
[fields[c] for c in cols],
|
|
)
|
|
trade_id = int(cur.lastrowid)
|
|
action = "inserted"
|
|
_mark_option_legs_excluded(conn, plan_id, legs)
|
|
del trade_id
|
|
return action
|
|
|
|
|
|
def _mark_option_legs_excluded(
|
|
conn: sqlite3.Connection,
|
|
plan_id: int,
|
|
legs: list[dict[str, Any]],
|
|
) -> int:
|
|
"""纯期权记录若 inst_id 出现在对冲腿中,标记排除以免双计."""
|
|
inst_ids = {
|
|
str(leg.get("inst_id") or "").strip()
|
|
for leg in legs
|
|
if str(leg.get("leg_role") or "").startswith("option") and str(leg.get("inst_id") or "").strip()
|
|
}
|
|
if not inst_ids:
|
|
return 0
|
|
n = 0
|
|
for inst_id in inst_ids:
|
|
cur = conn.execute(
|
|
"""
|
|
UPDATE options_review_trades
|
|
SET excluded_as_hedge_leg = 1, linked_hedge_plan_id = ?
|
|
WHERE source_type = ? AND inst_id = ? AND excluded_as_hedge_leg = 0
|
|
""",
|
|
(plan_id, SOURCE_OPTION, inst_id),
|
|
)
|
|
n += int(cur.rowcount or 0)
|
|
return n
|
|
|
|
|
|
def sync_hedge_plans_closed(conn: sqlite3.Connection) -> dict[str, Any]:
|
|
"""从本地 hedge_plans 导入已结束计划(计划级)."""
|
|
init_options_review_tables(conn)
|
|
from lib.hedge_plan.hedge_plan_db import get_plan_legs, init_hedge_plan_tables, list_plans
|
|
|
|
init_hedge_plan_tables(conn)
|
|
plans = list_plans(conn, status="closed", limit=500)
|
|
inserted = updated = skipped = 0
|
|
for plan in plans:
|
|
legs = get_plan_legs(conn, int(plan["id"]))
|
|
action = upsert_hedge_plan_row(conn, plan, legs)
|
|
if action == "inserted":
|
|
inserted += 1
|
|
elif action == "updated":
|
|
updated += 1
|
|
else:
|
|
skipped += 1
|
|
last_id = max((int(p["id"]) for p in plans), default=0)
|
|
set_sync_state(conn, "hedge_last_sync_at", _now_str())
|
|
set_sync_state(conn, "hedge_last_plan_id", str(last_id))
|
|
return {
|
|
"ok": True,
|
|
"fetched": len(plans),
|
|
"inserted": inserted,
|
|
"updated": updated,
|
|
"skipped": skipped,
|
|
}
|
|
|
|
|
|
def sync_all_review_sources(
|
|
conn: sqlite3.Connection,
|
|
ex: Any | None = None,
|
|
*,
|
|
options_limit: int = 500,
|
|
from_exchange: bool = False,
|
|
fetch_fn: Optional[Callable[..., list[dict[str, Any]]]] = None,
|
|
format_fn: Optional[Callable[..., dict[str, Any]]] = None,
|
|
) -> dict[str, Any]:
|
|
"""默认只读本地 options_trades + 已结束对冲计划;不访问交易所."""
|
|
init_options_review_tables(conn)
|
|
out: dict[str, Any] = {"ok": True, "options": None, "hedge": None}
|
|
if from_exchange and ex is not None:
|
|
out["options"] = sync_options_from_exchange(
|
|
conn, ex, limit=options_limit, fetch_fn=fetch_fn, format_fn=format_fn
|
|
)
|
|
else:
|
|
out["options"] = sync_options_from_local_trades(conn)
|
|
out["hedge"] = sync_hedge_plans_closed(conn)
|
|
return out
|
|
|
|
|
|
def ensure_local_review_synced(conn: sqlite3.Connection) -> dict[str, Any]:
|
|
"""列表/统计前轻量刷新本地源."""
|
|
return sync_all_review_sources(conn, from_exchange=False)
|
|
|
|
|
|
def _row_to_dict(row: Any) -> dict[str, Any]:
|
|
return dict(row) if row is not None else {}
|
|
|
|
|
|
def enrich_trade_row(row: dict[str, Any], entry: dict[str, Any] | None = None) -> dict[str, Any]:
|
|
out = dict(row)
|
|
out["source_label"] = SOURCE_LABELS.get(str(out.get("source_type") or ""), out.get("source_type"))
|
|
out["is_hedge"] = str(out.get("source_type") or "") in (SOURCE_PERP_OPTIONS, SOURCE_OPTIONS_OPTIONS)
|
|
legs = []
|
|
if out.get("legs_json"):
|
|
try:
|
|
legs = json.loads(str(out["legs_json"]))
|
|
except (TypeError, ValueError, json.JSONDecodeError):
|
|
legs = []
|
|
out["legs"] = legs if isinstance(legs, list) else []
|
|
out["reviewed"] = bool(entry)
|
|
if entry:
|
|
out["entry"] = dict(entry)
|
|
out["entry"]["images"] = parse_options_review_images_json(entry.get("images_json"))
|
|
out["strategy_tag"] = entry.get("strategy_tag")
|
|
out["result_tag"] = entry.get("result_tag")
|
|
out["reviewed_at"] = entry.get("reviewed_at") or entry.get("updated_at")
|
|
else:
|
|
out["entry"] = None
|
|
out["strategy_tag"] = None
|
|
out["result_tag"] = None
|
|
out["reviewed_at"] = None
|
|
return out
|
|
|
|
|
|
def _review_search_tokens(q: str) -> list[str]:
|
|
"""自由搜索词:BTCUSDT 同时匹配 BTC / BTCUSDT."""
|
|
raw = str(q or "").strip()
|
|
if not raw:
|
|
return []
|
|
tokens = [raw]
|
|
u = raw.upper()
|
|
for suf in ("-USDT", "-USD", "-USDC", "USDT", "USD", "USDC"):
|
|
if u.endswith(suf) and len(u) > len(suf):
|
|
base = u[: -len(suf)].rstrip("-_")
|
|
if base and base not in {t.upper() for t in tokens}:
|
|
tokens.append(base)
|
|
break
|
|
return tokens
|
|
|
|
|
|
def _review_trades_filters(
|
|
*,
|
|
source_type: str | None = None,
|
|
underlying: str | None = None,
|
|
opt_type: str | None = None,
|
|
strategy_tag: str | None = None,
|
|
q: str | None = None,
|
|
reviewed: str | None = None,
|
|
include_hedge_legs: bool = False,
|
|
closed_from: str | None = None,
|
|
closed_to: str | None = None,
|
|
) -> tuple[str, list[Any]]:
|
|
wheres: list[str] = []
|
|
args: list[Any] = []
|
|
if source_type and source_type in SOURCE_TYPES:
|
|
wheres.append("t.source_type=?")
|
|
args.append(source_type)
|
|
if underlying:
|
|
wheres.append("UPPER(COALESCE(t.underlying,''))=?")
|
|
args.append(underlying.strip().upper())
|
|
if opt_type:
|
|
ot = opt_type.strip().upper()
|
|
if ot in ("C", "P", "CALL", "PUT"):
|
|
if ot.startswith("C"):
|
|
ot = "C"
|
|
elif ot.startswith("P"):
|
|
ot = "P"
|
|
wheres.append(
|
|
"""(
|
|
UPPER(COALESCE(t.opt_type,''))=?
|
|
OR (
|
|
t.legs_json IS NOT NULL
|
|
AND t.legs_json LIKE '%' || '"opt_type":"' || ? || '%'
|
|
)
|
|
)"""
|
|
)
|
|
args.extend([ot, ot])
|
|
if not include_hedge_legs:
|
|
wheres.append("COALESCE(t.excluded_as_hedge_leg,0)=0")
|
|
if closed_from:
|
|
wheres.append("COALESCE(t.closed_at,'')>=?")
|
|
args.append(closed_from)
|
|
if closed_to:
|
|
wheres.append("COALESCE(t.closed_at,'')<=?")
|
|
args.append(closed_to)
|
|
# 兼容旧参数:精确策略标签;前端已改用 q 模糊搜索
|
|
if strategy_tag and not q:
|
|
wheres.append("UPPER(COALESCE(e.strategy_tag,''))=UPPER(?)")
|
|
args.append(strategy_tag)
|
|
search_tokens = _review_search_tokens(q or "")
|
|
if search_tokens:
|
|
token_ors: list[str] = []
|
|
for tok in search_tokens:
|
|
like = f"%{tok}%"
|
|
token_ors.append(
|
|
"""(
|
|
UPPER(COALESCE(t.underlying,'')) LIKE UPPER(?)
|
|
OR UPPER(COALESCE(t.inst_id,'')) LIKE UPPER(?)
|
|
OR UPPER(COALESCE(t.legs_json,'')) LIKE UPPER(?)
|
|
OR UPPER(COALESCE(e.strategy_tag,'')) LIKE UPPER(?)
|
|
OR UPPER(COALESCE(e.result_tag,'')) LIKE UPPER(?)
|
|
)"""
|
|
)
|
|
args.extend([like, like, like, like, like])
|
|
wheres.append("(" + " OR ".join(token_ors) + ")")
|
|
if reviewed == "1" or reviewed == "yes":
|
|
wheres.append("e.id IS NOT NULL")
|
|
elif reviewed == "0" or reviewed == "no":
|
|
wheres.append("e.id IS NULL")
|
|
where = (" WHERE " + " AND ".join(wheres)) if wheres else ""
|
|
return where, args
|
|
|
|
|
|
def count_review_trades(
|
|
conn: sqlite3.Connection,
|
|
*,
|
|
source_type: str | None = None,
|
|
underlying: str | None = None,
|
|
opt_type: str | None = None,
|
|
strategy_tag: str | None = None,
|
|
q: str | None = None,
|
|
reviewed: str | None = None,
|
|
include_hedge_legs: bool = False,
|
|
closed_from: str | None = None,
|
|
closed_to: str | None = None,
|
|
) -> int:
|
|
init_options_review_tables(conn)
|
|
where, args = _review_trades_filters(
|
|
source_type=source_type,
|
|
underlying=underlying,
|
|
opt_type=opt_type,
|
|
strategy_tag=strategy_tag,
|
|
q=q,
|
|
reviewed=reviewed,
|
|
include_hedge_legs=include_hedge_legs,
|
|
closed_from=closed_from,
|
|
closed_to=closed_to,
|
|
)
|
|
row = conn.execute(
|
|
f"""
|
|
SELECT COUNT(*) AS c
|
|
FROM options_review_trades t
|
|
LEFT JOIN options_review_entries e ON e.trade_id = t.id
|
|
{where}
|
|
""",
|
|
args,
|
|
).fetchone()
|
|
return int(row["c"] if row else 0)
|
|
|
|
|
|
def list_review_trades(
|
|
conn: sqlite3.Connection,
|
|
*,
|
|
source_type: str | None = None,
|
|
underlying: str | None = None,
|
|
opt_type: str | None = None,
|
|
strategy_tag: str | None = None,
|
|
q: str | None = None,
|
|
reviewed: str | None = None,
|
|
include_hedge_legs: bool = False,
|
|
closed_from: str | None = None,
|
|
closed_to: str | None = None,
|
|
limit: int = 200,
|
|
offset: int = 0,
|
|
) -> list[dict[str, Any]]:
|
|
init_options_review_tables(conn)
|
|
where, args = _review_trades_filters(
|
|
source_type=source_type,
|
|
underlying=underlying,
|
|
opt_type=opt_type,
|
|
strategy_tag=strategy_tag,
|
|
q=q,
|
|
reviewed=reviewed,
|
|
include_hedge_legs=include_hedge_legs,
|
|
closed_from=closed_from,
|
|
closed_to=closed_to,
|
|
)
|
|
rows = conn.execute(
|
|
f"""
|
|
SELECT t.*, e.id AS entry_id, e.strategy_tag AS e_strategy_tag,
|
|
e.direction_view, e.entry_logic, e.exit_reason, e.followed_plan,
|
|
e.mistake_tags, e.result_tag, e.note, e.images_json, e.image,
|
|
e.reviewed_at, e.updated_at
|
|
FROM options_review_trades t
|
|
LEFT JOIN options_review_entries e ON e.trade_id = t.id
|
|
{where}
|
|
ORDER BY COALESCE(t.closed_at, t.opened_at, '') DESC, t.id DESC
|
|
LIMIT ? OFFSET ?
|
|
""",
|
|
[*args, int(limit), int(offset)],
|
|
).fetchall()
|
|
out: list[dict[str, Any]] = []
|
|
for r in rows:
|
|
d = _row_to_dict(r)
|
|
entry = None
|
|
if d.get("entry_id"):
|
|
entry = {
|
|
"id": d.pop("entry_id", None),
|
|
"strategy_tag": d.pop("e_strategy_tag", None),
|
|
"direction_view": d.pop("direction_view", None),
|
|
"entry_logic": d.pop("entry_logic", None),
|
|
"exit_reason": d.pop("exit_reason", None),
|
|
"followed_plan": d.pop("followed_plan", None),
|
|
"mistake_tags": d.pop("mistake_tags", None),
|
|
"result_tag": d.pop("result_tag", None),
|
|
"note": d.pop("note", None),
|
|
"images_json": d.pop("images_json", None),
|
|
"image": d.pop("image", None),
|
|
"reviewed_at": d.pop("reviewed_at", None),
|
|
"updated_at": d.pop("updated_at", None),
|
|
}
|
|
else:
|
|
for k in (
|
|
"entry_id",
|
|
"e_strategy_tag",
|
|
"direction_view",
|
|
"entry_logic",
|
|
"exit_reason",
|
|
"followed_plan",
|
|
"mistake_tags",
|
|
"result_tag",
|
|
"note",
|
|
"images_json",
|
|
"image",
|
|
"reviewed_at",
|
|
"updated_at",
|
|
):
|
|
d.pop(k, None)
|
|
out.append(enrich_trade_row(d, entry))
|
|
return out
|
|
|
|
|
|
def get_review_trade(conn: sqlite3.Connection, trade_id: int) -> dict[str, Any] | None:
|
|
init_options_review_tables(conn)
|
|
row = conn.execute(
|
|
"SELECT * FROM options_review_trades WHERE id=?", (int(trade_id),)
|
|
).fetchone()
|
|
if not row:
|
|
return None
|
|
entry_row = conn.execute(
|
|
"SELECT * FROM options_review_entries WHERE trade_id=?", (int(trade_id),)
|
|
).fetchone()
|
|
entry = _row_to_dict(entry_row) if entry_row else None
|
|
return enrich_trade_row(_row_to_dict(row), entry)
|
|
|
|
|
|
def save_review_entry(
|
|
conn: sqlite3.Connection,
|
|
trade_id: int,
|
|
payload: dict[str, Any],
|
|
) -> dict[str, Any]:
|
|
"""保存/更新人工复盘;不影响 trades 快照字段."""
|
|
init_options_review_tables(conn)
|
|
trade = conn.execute(
|
|
"SELECT id FROM options_review_trades WHERE id=?", (int(trade_id),)
|
|
).fetchone()
|
|
if not trade:
|
|
return {"ok": False, "msg": "交易不存在"}
|
|
images = payload.get("images")
|
|
if images is None and payload.get("images_json") is not None:
|
|
images = parse_options_review_images_json(payload.get("images_json"))
|
|
if not isinstance(images, list):
|
|
images = []
|
|
images_json = images_json_dumps(images)
|
|
primary = None
|
|
if images:
|
|
primary = str(images[0].get("file") or "").strip() or None
|
|
fields = {
|
|
"strategy_tag": str(payload.get("strategy_tag") or "").strip() or None,
|
|
"direction_view": str(payload.get("direction_view") or "").strip() or None,
|
|
"entry_logic": str(payload.get("entry_logic") or "").strip() or None,
|
|
"exit_reason": str(payload.get("exit_reason") or "").strip() or None,
|
|
"followed_plan": str(payload.get("followed_plan") or "").strip() or None,
|
|
"mistake_tags": str(payload.get("mistake_tags") or "").strip() or None,
|
|
"result_tag": str(payload.get("result_tag") or "").strip() or None,
|
|
"note": str(payload.get("note") or "").strip() or None,
|
|
"images_json": images_json,
|
|
"image": primary or (str(payload.get("image") or "").strip() or None),
|
|
"updated_at": _now_str(),
|
|
}
|
|
existing = conn.execute(
|
|
"SELECT id, reviewed_at FROM options_review_entries WHERE trade_id=?",
|
|
(int(trade_id),),
|
|
).fetchone()
|
|
if existing:
|
|
sets = ", ".join(f"{k}=?" for k in fields)
|
|
conn.execute(
|
|
f"UPDATE options_review_entries SET {sets} WHERE trade_id=?",
|
|
[*fields.values(), int(trade_id)],
|
|
)
|
|
else:
|
|
fields["trade_id"] = int(trade_id)
|
|
fields["reviewed_at"] = _now_str()
|
|
cols = list(fields.keys())
|
|
conn.execute(
|
|
f"INSERT INTO options_review_entries ({','.join(cols)}) VALUES ({','.join(['?']*len(cols))})",
|
|
[fields[c] for c in cols],
|
|
)
|
|
return {"ok": True, "trade": get_review_trade(conn, int(trade_id))}
|
|
|
|
|
|
def delete_review_entry(conn: sqlite3.Connection, trade_id: int) -> dict[str, Any]:
|
|
init_options_review_tables(conn)
|
|
entry = conn.execute(
|
|
"SELECT * FROM options_review_entries WHERE trade_id=?", (int(trade_id),)
|
|
).fetchone()
|
|
if not entry:
|
|
return {"ok": False, "msg": "无复盘记录"}
|
|
conn.execute("DELETE FROM options_review_entries WHERE trade_id=?", (int(trade_id),))
|
|
return {"ok": True, "entry": _row_to_dict(entry)}
|
|
|
|
|
|
def _hold_bucket(sec: Optional[int]) -> str:
|
|
if sec is None:
|
|
return "未知"
|
|
for label, lo, hi in HOLD_BUCKETS:
|
|
if sec >= lo and (hi is None or sec < hi):
|
|
return label
|
|
return "未知"
|
|
|
|
|
|
def _group_stats(rows: list[dict[str, Any]], key_fn) -> list[dict[str, Any]]:
|
|
buckets: dict[str, dict[str, Any]] = {}
|
|
for row in rows:
|
|
key = str(key_fn(row) or "未填")
|
|
b = buckets.setdefault(
|
|
key,
|
|
{"key": key, "count": 0, "wins": 0, "losses": 0, "pnl_sum": 0.0, "hold_sum": 0.0, "hold_n": 0},
|
|
)
|
|
pnl = _safe_float(row.get("realized_pnl_total"))
|
|
if pnl is None:
|
|
continue
|
|
b["count"] += 1
|
|
b["pnl_sum"] = round(b["pnl_sum"] + pnl, 4)
|
|
if pnl > 0:
|
|
b["wins"] += 1
|
|
elif pnl < 0:
|
|
b["losses"] += 1
|
|
hs = row.get("hold_seconds")
|
|
if hs is not None:
|
|
try:
|
|
b["hold_sum"] += float(hs)
|
|
b["hold_n"] += 1
|
|
except (TypeError, ValueError):
|
|
pass
|
|
out = []
|
|
for b in buckets.values():
|
|
c = b["count"]
|
|
out.append(
|
|
{
|
|
"key": b["key"],
|
|
"count": c,
|
|
"wins": b["wins"],
|
|
"losses": b["losses"],
|
|
"win_rate": round(b["wins"] / c * 100, 2) if c else 0,
|
|
"pnl_sum": round(b["pnl_sum"], 4),
|
|
"avg_pnl": round(b["pnl_sum"] / c, 4) if c else None,
|
|
"avg_hold_sec": round(b["hold_sum"] / b["hold_n"], 1) if b["hold_n"] else None,
|
|
}
|
|
)
|
|
out.sort(key=lambda x: abs(float(x.get("pnl_sum") or 0)), reverse=True)
|
|
return out
|
|
|
|
|
|
def compute_review_stats(
|
|
conn: sqlite3.Connection,
|
|
*,
|
|
source_type: str | None = None,
|
|
underlying: str | None = None,
|
|
include_hedge_legs: bool = False,
|
|
closed_from: str | None = None,
|
|
closed_to: str | None = None,
|
|
require_strategy: bool = False,
|
|
) -> dict[str, Any]:
|
|
rows = list_review_trades(
|
|
conn,
|
|
source_type=source_type,
|
|
underlying=underlying,
|
|
include_hedge_legs=include_hedge_legs,
|
|
closed_from=closed_from,
|
|
closed_to=closed_to,
|
|
limit=5000,
|
|
offset=0,
|
|
)
|
|
if require_strategy:
|
|
rows = [r for r in rows if str(r.get("strategy_tag") or "").strip()]
|
|
|
|
wins = losses = reviewed = 0
|
|
pnl_sum = 0.0
|
|
hold_vals: list[float] = []
|
|
for r in rows:
|
|
if r.get("reviewed"):
|
|
reviewed += 1
|
|
pnl = _safe_float(r.get("realized_pnl_total"))
|
|
if pnl is None:
|
|
continue
|
|
pnl_sum += pnl
|
|
if pnl > 0:
|
|
wins += 1
|
|
elif pnl < 0:
|
|
losses += 1
|
|
if r.get("hold_seconds") is not None:
|
|
hold_vals.append(float(r["hold_seconds"]))
|
|
|
|
total = wins + losses
|
|
kpi = {
|
|
"total": len(rows),
|
|
"pnl_count": total,
|
|
"reviewed": reviewed,
|
|
"review_rate": round(reviewed / len(rows) * 100, 2) if rows else 0,
|
|
"wins": wins,
|
|
"losses": losses,
|
|
"win_rate": round(wins / total * 100, 2) if total else 0,
|
|
"pnl_sum": round(pnl_sum, 4),
|
|
"avg_pnl": round(pnl_sum / total, 4) if total else None,
|
|
"avg_hold_sec": round(sum(hold_vals) / len(hold_vals), 1) if hold_vals else None,
|
|
}
|
|
|
|
strategy_rows = [r for r in rows if str(r.get("strategy_tag") or "").strip()]
|
|
return {
|
|
"ok": True,
|
|
"kpi": kpi,
|
|
"by_source_type": _group_stats(rows, lambda r: SOURCE_LABELS.get(str(r.get("source_type") or ""), r.get("source_type"))),
|
|
"by_underlying": _group_stats(rows, lambda r: r.get("underlying") or "未填"),
|
|
"by_opt_type": _group_stats(
|
|
[r for r in rows if r.get("source_type") == SOURCE_OPTION],
|
|
lambda r: r.get("opt_type") or "未填",
|
|
),
|
|
"by_strategy": _group_stats(strategy_rows, lambda r: r.get("strategy_tag")),
|
|
"by_close_reason": _group_stats(
|
|
[r for r in rows if r.get("is_hedge")],
|
|
lambda r: r.get("plan_close_reason") or "未填",
|
|
),
|
|
"by_hold_bucket": _group_stats(rows, lambda r: _hold_bucket(r.get("hold_seconds"))),
|
|
"sync": {
|
|
"options_last_sync_at": get_sync_state(conn, "options_last_sync_at"),
|
|
"hedge_last_sync_at": get_sync_state(conn, "hedge_last_sync_at"),
|
|
"hedge_last_plan_id": get_sync_state(conn, "hedge_last_plan_id"),
|
|
},
|
|
}
|