dekun
1396 lines
50 KiB
Python
1396 lines
50 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]:
|
||
try:
|
||
row = conn.execute(
|
||
"SELECT value FROM options_review_sync_state WHERE key=?", (key,)
|
||
).fetchone()
|
||
except sqlite3.OperationalError:
|
||
return None
|
||
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:
|
||
try:
|
||
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()),
|
||
)
|
||
except sqlite3.OperationalError:
|
||
# 列表/统计并发刷新时可能撞锁;标记失败不阻断读列表
|
||
pass
|
||
|
||
|
||
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,
|
||
"initial_premium_usdt": _safe_float(row.get("initial_premium_usdt")),
|
||
"profit_rate": _safe_float(row.get("profit_rate")),
|
||
"option_fee_usdt": _safe_float(row.get("option_fee_usdt")),
|
||
}
|
||
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 _sum_sim_order_fees(conn: sqlite3.Connection, ord_ids: list[str]) -> float | None:
|
||
"""从 sim_option_orders 汇总手续费(权利金币种,正数)."""
|
||
ids = [str(x).strip() for x in (ord_ids or []) if str(x).strip()]
|
||
if not ids:
|
||
return None
|
||
try:
|
||
conn.execute("SELECT 1 FROM sim_option_orders LIMIT 1").fetchone()
|
||
except Exception:
|
||
return None
|
||
total = 0.0
|
||
n = 0
|
||
for oid in ids:
|
||
try:
|
||
row = conn.execute(
|
||
"SELECT fee FROM sim_option_orders WHERE ord_id=?",
|
||
(oid,),
|
||
).fetchone()
|
||
except Exception:
|
||
continue
|
||
if not row:
|
||
continue
|
||
fee = _safe_float(row["fee"] if hasattr(row, "keys") else row[0])
|
||
if fee is None:
|
||
continue
|
||
total += abs(float(fee))
|
||
n += 1
|
||
return round(total, 8) if n else None
|
||
|
||
|
||
def _estimate_okx_option_fee_coin(
|
||
*,
|
||
underlying_coin: float | None,
|
||
premium_coin: float | None,
|
||
) -> float:
|
||
"""按 OKX 币本位公式估算期权手续费(权利金币种)."""
|
||
try:
|
||
from lib.sim.pricing_lib import option_fee_coin
|
||
|
||
return float(
|
||
option_fee_coin(
|
||
underlying_coin=float(underlying_coin or 0.0),
|
||
premium_coin=float(premium_coin or 0.0),
|
||
)
|
||
)
|
||
except Exception:
|
||
return 0.0
|
||
|
||
|
||
def _fee_looks_like_legacy_premium_rate(
|
||
fee_coin: float,
|
||
*,
|
||
premium_coin: float | None,
|
||
underlying_coin: float | None,
|
||
) -> bool:
|
||
"""旧模拟盘把手续费算成权利金×费率,约为正确值的 premium/underlying 倍."""
|
||
if fee_coin <= 0:
|
||
return True
|
||
prem = abs(float(premium_coin or 0.0))
|
||
uly = abs(float(underlying_coin or 0.0))
|
||
if prem <= 0 or uly <= 0:
|
||
return False
|
||
# 正确量级 ~ uly * fee_rate; 错误量级 ~ prem * fee_rate
|
||
# 若 fee << uly*1e-4 且接近 prem*5e-4,视为旧算法
|
||
try:
|
||
from lib.sim.pricing_lib import sim_fee_rate
|
||
|
||
fr = float(sim_fee_rate())
|
||
except Exception:
|
||
fr = 0.0005
|
||
expect_okx = uly * fr
|
||
expect_legacy = prem * fr
|
||
if expect_okx <= 0:
|
||
return False
|
||
# 比 OKX 估算小一个数量级以上,且靠近旧公式
|
||
if fee_coin < expect_okx * 0.25 and abs(fee_coin - expect_legacy) <= max(expect_legacy * 0.5, 1e-10):
|
||
return True
|
||
if fee_coin < expect_okx * 0.05:
|
||
return True
|
||
return False
|
||
|
||
|
||
def _resolve_option_fees_coin(
|
||
conn: sqlite3.Connection,
|
||
row: Any,
|
||
*,
|
||
open_fee: float,
|
||
close_fee: float,
|
||
premium_ccy: str,
|
||
) -> tuple[float, float]:
|
||
"""解析期权开/平手续费(权利金币种);修正模拟盘旧算法偏小值."""
|
||
ccy = (premium_ccy or "").strip().upper()
|
||
paid = _safe_float(row["premium_paid"] if hasattr(row, "keys") else None)
|
||
recv = _safe_float(row["premium_received"] if hasattr(row, "keys") else None)
|
||
uly = _safe_float(row["eth_amount"] if hasattr(row, "keys") else None)
|
||
if uly is None or uly <= 0:
|
||
sheets = _safe_float(row["sheets"] if hasattr(row, "keys") else None) or 0.0
|
||
# BTC 期权约 0.01,ETH 约 0.1
|
||
ct = 0.01 if ccy == "BTC" else 0.1
|
||
uly = float(sheets) * ct if sheets > 0 else None
|
||
|
||
open_oids = [
|
||
p.strip()
|
||
for p in str(row["exchange_ord_id"] if hasattr(row, "keys") else "").split(",")
|
||
if p.strip()
|
||
]
|
||
close_oids = [
|
||
p.strip()
|
||
for p in str(row["close_ord_id"] if hasattr(row, "keys") else "").split(",")
|
||
if p.strip()
|
||
]
|
||
|
||
ofee = abs(float(open_fee or 0.0))
|
||
cfee = abs(float(close_fee or 0.0))
|
||
|
||
# 优先用订单表(若订单本身仍是旧算法,下面会再纠正)
|
||
sim_open = _sum_sim_order_fees(conn, open_oids)
|
||
sim_close = _sum_sim_order_fees(conn, close_oids)
|
||
if sim_open is not None and sim_open > 0:
|
||
ofee = sim_open
|
||
if sim_close is not None and sim_close > 0:
|
||
cfee = sim_close
|
||
|
||
if ccy in ("ETH", "BTC"):
|
||
if _fee_looks_like_legacy_premium_rate(ofee, premium_coin=paid, underlying_coin=uly):
|
||
est = _estimate_okx_option_fee_coin(underlying_coin=uly, premium_coin=paid)
|
||
if est > 0:
|
||
ofee = est
|
||
if _fee_looks_like_legacy_premium_rate(cfee, premium_coin=recv, underlying_coin=uly):
|
||
est = _estimate_okx_option_fee_coin(underlying_coin=uly, premium_coin=recv)
|
||
if est > 0:
|
||
cfee = est
|
||
elif cfee <= 0 and recv is not None and recv > 0:
|
||
est = _estimate_okx_option_fee_coin(underlying_coin=uly, premium_coin=recv)
|
||
if est > 0:
|
||
cfee = est
|
||
if ofee <= 0 and paid is not None and paid > 0:
|
||
est = _estimate_okx_option_fee_coin(underlying_coin=uly, premium_coin=paid)
|
||
if est > 0:
|
||
ofee = est
|
||
|
||
return round(ofee, 8), round(cfee, 8)
|
||
|
||
|
||
def sync_options_from_local_trades(
|
||
conn: sqlite3.Connection,
|
||
ex: Any | None = None,
|
||
) -> dict[str, Any]:
|
||
"""从本地 options_trades 已平仓记录导入复盘快照(不访问交易所).
|
||
|
||
盈亏优先用已冻结 realized_pnl_usdt(模拟盘现金口径 / 平仓时指数),
|
||
禁止每次同步用实时指数重算,否则复盘数字会跟盘跳动.
|
||
"""
|
||
init_options_review_tables(conn)
|
||
from lib.options.options_db import init_options_tables
|
||
from lib.options.options_margin_mode_lib import margin_mode_from_inst_id, premium_ccy_for_mode
|
||
|
||
init_options_tables(conn)
|
||
# 模拟盘先把现金盈亏写进 realized_pnl_usdt,复盘与持仓统计同一口径.
|
||
# 不可 is_sim_mode(lambda: conn): with_db 会 close 掉调用方连接.
|
||
try:
|
||
from lib.instance.runtime_settings_lib import ensure_runtime_settings_table, runtime_get
|
||
from lib.sim.mode_lib import MODE_SIM, default_trading_mode
|
||
|
||
ensure_runtime_settings_table(conn)
|
||
mode = runtime_get(conn, "trading.mode")
|
||
mode_s = str(mode or "").strip().lower() or default_trading_mode()
|
||
if mode_s == MODE_SIM:
|
||
from lib.options.options_stats_lib import sync_sim_cash_realized_pnl_usdt
|
||
|
||
sync_sim_cash_realized_pnl_usdt(conn, ex)
|
||
try:
|
||
conn.commit()
|
||
except Exception:
|
||
pass
|
||
except Exception:
|
||
pass
|
||
rows = conn.execute(
|
||
"""
|
||
SELECT id, inst_id, underlying, opt_type, strike, exp_time, sheets,
|
||
eth_amount, open_quote, close_quote, premium_paid, premium_received,
|
||
realized_pnl, realized_pnl_usdt, close_index_px, open_index_px,
|
||
budget_usdt, premium_ccy, created_at, closed_at, signal_note, status,
|
||
trading_usdt_before, pnl_settled, initial_premium_usdt,
|
||
open_fee, close_fee, exchange_ord_id, close_ord_id
|
||
FROM options_trades
|
||
WHERE status = 'closed'
|
||
ORDER BY id DESC
|
||
LIMIT 500
|
||
"""
|
||
).fetchall()
|
||
# 读完先提交,后续拉指数不与写事务叠在一起
|
||
try:
|
||
conn.commit()
|
||
except Exception:
|
||
pass
|
||
|
||
def _index_px(underly: str) -> float | None:
|
||
u = (underly or "ETH").strip().upper() or "ETH"
|
||
inst = f"{u}-USDT"
|
||
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)
|
||
for pub in pubs:
|
||
try:
|
||
if hasattr(pub, "public_get_market_ticker"):
|
||
tick_rows = (pub.public_get_market_ticker({"instId": inst}) or {}).get("data") or []
|
||
if tick_rows:
|
||
last = tick_rows[0].get("last") or tick_rows[0].get("lastPx")
|
||
if last is not None and float(last) > 0:
|
||
return float(last)
|
||
except Exception:
|
||
pass
|
||
try:
|
||
from lib.exchange.okx_options_lib import fetch_index_price
|
||
|
||
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 = t.get("last") or t.get("close")
|
||
if last is not None and float(last) > 0:
|
||
return float(last)
|
||
except Exception:
|
||
continue
|
||
return None
|
||
|
||
def _to_usdt(amount: float | None, *, ccy: str, idx: float | None) -> float | None:
|
||
if amount is None:
|
||
return None
|
||
unit = (ccy or "USDC").strip().upper()
|
||
if unit in ("ETH", "BTC"):
|
||
if idx is None or idx <= 0:
|
||
return None
|
||
return round(float(amount) * float(idx), 4)
|
||
return round(float(amount), 4)
|
||
|
||
inserted = updated = skipped = 0
|
||
idx_cache: dict[str, float | None] = {}
|
||
need_idx: set[str] = set()
|
||
for r in rows:
|
||
# 已有冻结 U 或平仓指数时不再打实时行情
|
||
if _safe_float(r["realized_pnl_usdt"]) is not None:
|
||
continue
|
||
frozen_idx = _safe_float(r["close_index_px"]) or _safe_float(r["open_index_px"])
|
||
if frozen_idx is not None and frozen_idx > 0:
|
||
continue
|
||
inst0 = str(r["inst_id"] or "")
|
||
underly0 = str(r["underlying"] or (inst0.split("-")[0] if inst0 else "ETH") or "ETH")
|
||
ccy0 = str(r["premium_ccy"] or "").strip().upper()
|
||
if not ccy0:
|
||
ccy0 = premium_ccy_for_mode(margin_mode_from_inst_id(inst0), underly0)
|
||
if ccy0 in ("ETH", "BTC"):
|
||
need_idx.add(underly0)
|
||
for underly0 in need_idx:
|
||
idx_cache[underly0] = _index_px(underly0)
|
||
|
||
for r in rows:
|
||
trade_id = int(r["id"])
|
||
history_key = f"local_opt:{trade_id}"
|
||
inst = str(r["inst_id"] or "")
|
||
underly = str(r["underlying"] or (inst.split("-")[0] if inst else "ETH") or "ETH")
|
||
ccy = str(r["premium_ccy"] or "").strip().upper()
|
||
if not ccy:
|
||
ccy = premium_ccy_for_mode(margin_mode_from_inst_id(inst), underly)
|
||
frozen_usdt = _safe_float(r["realized_pnl_usdt"])
|
||
try:
|
||
settled = int(r["pnl_settled"] or 0) == 1
|
||
except Exception:
|
||
settled = False
|
||
# 有账户快照未结算:交易记录显示待结算(真盈亏=卖回后 after−before)
|
||
acct_pending = _safe_float(r["trading_usdt_before"]) is not None and not settled
|
||
close_idx = _safe_float(r["close_index_px"])
|
||
open_idx = _safe_float(r["open_index_px"])
|
||
# 折算指数:仅用于初始权利金/期权手续费展示,不用于真盈亏
|
||
idx = None
|
||
if close_idx is not None and close_idx > 0:
|
||
idx = close_idx
|
||
elif open_idx is not None and open_idx > 0:
|
||
idx = open_idx
|
||
else:
|
||
idx = idx_cache.get(underly)
|
||
prem = _safe_float(r["premium_paid"])
|
||
# 交易记录真盈亏=口径 A(卖光币后账户差分);未结算不写假 U
|
||
if acct_pending:
|
||
pnl_u = None
|
||
elif frozen_usdt is not None and (settled or _safe_float(r["trading_usdt_before"]) is None):
|
||
pnl_u = round(float(frozen_usdt), 4)
|
||
else:
|
||
# 旧单无账户快照:不回退币×价冒充真盈亏
|
||
pnl_u = None
|
||
# 初始权利金(U)=开仓指数×权利金币量;不含多买
|
||
stored_initial = _safe_float(r["initial_premium_usdt"])
|
||
open_idx_for_prem = open_idx if (open_idx is not None and open_idx > 0) else idx
|
||
prem_u = stored_initial
|
||
if prem_u is None:
|
||
prem_u = _to_usdt(prem, ccy=ccy, idx=open_idx_for_prem)
|
||
if ccy in ("ETH", "BTC") and prem_u is None:
|
||
prem_u = prem
|
||
# 仅期权开/平手续费折 U(不含现货)
|
||
ofee = abs(_safe_float(r["open_fee"]) or 0.0)
|
||
cfee = abs(_safe_float(r["close_fee"]) or 0.0)
|
||
# 模拟盘旧单曾按「权利金×费率」计费偏小:按 OKX 公式/订单回填
|
||
ofee, cfee = _resolve_option_fees_coin(
|
||
conn,
|
||
r,
|
||
open_fee=ofee,
|
||
close_fee=cfee,
|
||
premium_ccy=ccy,
|
||
)
|
||
fee_u = None
|
||
if ccy in ("ETH", "BTC"):
|
||
fee_parts = 0.0
|
||
has_fee = False
|
||
if ofee > 0 and open_idx_for_prem is not None and open_idx_for_prem > 0:
|
||
fee_parts += ofee * float(open_idx_for_prem)
|
||
has_fee = True
|
||
sell_idx = close_idx if (close_idx is not None and close_idx > 0) else idx
|
||
if cfee > 0 and sell_idx is not None and sell_idx > 0:
|
||
fee_parts += cfee * float(sell_idx)
|
||
has_fee = True
|
||
if has_fee:
|
||
fee_u = round(fee_parts, 4)
|
||
elif ofee > 0 or cfee > 0:
|
||
fee_u = round(ofee + cfee, 4)
|
||
profit_rate = None
|
||
if pnl_u is not None and prem_u is not None and abs(float(prem_u)) > 1e-12:
|
||
profit_rate = round(float(pnl_u) / float(prem_u), 6)
|
||
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": prem_u if prem_u is not None else prem,
|
||
"realized_pnl": pnl_u,
|
||
"initial_premium_usdt": prem_u,
|
||
"profit_rate": profit_rate,
|
||
"option_fee_usdt": fee_u,
|
||
"created_at": opened_at,
|
||
"closed_at": closed_at,
|
||
"status_label": "待结算" if acct_pending else "已平",
|
||
},
|
||
)
|
||
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")),
|
||
"profit_rr": _safe_float(plan.get("profit_rr")),
|
||
"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]:
|
||
"""对冲模块已移除:不再导入对冲计划到复盘."""
|
||
init_options_review_tables(conn)
|
||
return {"ok": True, "fetched": 0, "inserted": 0, "updated": 0, "skipped": 0, "removed": True}
|
||
|
||
|
||
|
||
|
||
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, ex=ex)
|
||
out["hedge"] = sync_hedge_plans_closed(conn)
|
||
return out
|
||
|
||
|
||
def ensure_local_review_synced(
|
||
conn: sqlite3.Connection,
|
||
*,
|
||
ex: Any | None = None,
|
||
backfill_exchange_pnl: bool = True,
|
||
) -> dict[str, Any]:
|
||
"""列表/统计前轻量刷新本地源;有交易所时先用历史仓位盈亏覆盖本地再导入复盘."""
|
||
from lib.options.options_time_lib import (
|
||
migrate_options_utc_wall_clock_to_app,
|
||
repair_swapped_closed_at_signal_note,
|
||
)
|
||
|
||
ts_migrate: dict[str, Any] = {}
|
||
note_swap: dict[str, Any] = {}
|
||
try:
|
||
ts_migrate = migrate_options_utc_wall_clock_to_app(conn)
|
||
except Exception as e:
|
||
ts_migrate = {"ok": False, "msg": str(e)}
|
||
try:
|
||
note_swap = repair_swapped_closed_at_signal_note(conn)
|
||
except Exception as e:
|
||
note_swap = {"ok": False, "msg": str(e)}
|
||
# 提交后再拉历史/行情,避免模拟盘 broker 二次开库与本连接互锁
|
||
try:
|
||
conn.commit()
|
||
except Exception:
|
||
pass
|
||
# 模拟盘用现金冻结,勿再用交易所历史+实时指数覆盖
|
||
sim_mode = False
|
||
try:
|
||
from lib.instance.runtime_settings_lib import ensure_runtime_settings_table, runtime_get
|
||
from lib.sim.mode_lib import MODE_SIM, default_trading_mode
|
||
|
||
ensure_runtime_settings_table(conn)
|
||
mode = runtime_get(conn, "trading.mode")
|
||
sim_mode = (str(mode or "").strip().lower() or default_trading_mode()) == MODE_SIM
|
||
except Exception:
|
||
sim_mode = False
|
||
if backfill_exchange_pnl and ex is not None and not sim_mode:
|
||
try:
|
||
from lib.exchange.okx_options_lib import fetch_all_option_positions_history
|
||
from lib.options.options_monitor_lib import (
|
||
backfill_closed_options_realized_pnl_from_history,
|
||
)
|
||
|
||
hist = fetch_all_option_positions_history(ex, limit=200)
|
||
backfill_closed_options_realized_pnl_from_history(conn, hist)
|
||
conn.commit()
|
||
except Exception:
|
||
try:
|
||
conn.rollback()
|
||
except Exception:
|
||
pass
|
||
try:
|
||
out = sync_all_review_sources(conn, ex=ex, from_exchange=False)
|
||
except sqlite3.OperationalError as e:
|
||
out = {"ok": False, "msg": f"sync locked: {e}"}
|
||
if isinstance(out, dict):
|
||
out["ts_migrate"] = ts_migrate
|
||
out["note_swap"] = note_swap
|
||
if out.get("ok") is not False:
|
||
out["ok"] = True
|
||
return out
|
||
|
||
|
||
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)
|
||
# 盈利率缺失时按 复盘盈亏/初始权利金 补算
|
||
if _safe_float(out.get("profit_rate")) is None:
|
||
pnl = _safe_float(out.get("realized_pnl_total"))
|
||
prem = _safe_float(out.get("initial_premium_usdt"))
|
||
if pnl is not None and prem is not None and abs(prem) > 1e-12:
|
||
out["profit_rate"] = round(pnl / prem, 6)
|
||
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["direction_view"] = entry.get("direction_view")
|
||
out["entry_logic"] = entry.get("entry_logic")
|
||
out["prev_entry_tf"] = entry.get("prev_entry_tf")
|
||
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["direction_view"] = None
|
||
out["entry_logic"] = None
|
||
out["prev_entry_tf"] = 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.prev_entry_tf, 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),
|
||
"prev_entry_tf": d.pop("prev_entry_tf", 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",
|
||
"prev_entry_tf",
|
||
"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
|
||
prev_tf = str(payload.get("prev_entry_tf") or "").strip().lower()
|
||
if prev_tf and prev_tf not in ("15m", "1h", "2h", "4h", "8h", "12h"):
|
||
prev_tf = ""
|
||
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,
|
||
"prev_entry_tf": prev_tf 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"),
|
||
},
|
||
}
|