dekun
1391 lines
50 KiB
Python
1391 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)
|
|
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)
|
|
pnl = _safe_float(r["realized_pnl"])
|
|
if pnl is None:
|
|
paid0 = _safe_float(r["premium_paid"])
|
|
recv0 = _safe_float(r["premium_received"])
|
|
if paid0 is not None and recv0 is not None:
|
|
pnl = recv0 - paid0
|
|
prem = _safe_float(r["premium_paid"])
|
|
# 期权交易记录盈亏与数据统计同口径 B:币盈亏 × 平仓卖回价(不用账户差分 A)
|
|
pnl_u = _to_usdt(pnl, ccy=ccy, idx=idx)
|
|
# 初始权利金(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 pnl_u is None:
|
|
pnl_u = None
|
|
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 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"),
|
|
},
|
|
}
|