dekun
48afc8ebe3
Drop exchange sync for the review page, auto-import closed local options and hedge plans, and match contract-style multi-timeframe upload styling with smaller fonts. Co-authored-by: Cursor <cursoragent@cursor.com>
311 lines
11 KiB
Python
311 lines
11 KiB
Python
"""期权复盘(含对冲)单元测试:导入去重、双计防护、复盘不被覆盖、统计."""
|
|
from __future__ import annotations
|
|
|
|
import sqlite3
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
from lib.hedge_plan.hedge_plan_db import init_hedge_plan_tables, insert_leg, insert_plan
|
|
from lib.options.options_review_db import SOURCE_OPTION, SOURCE_PERP_OPTIONS, init_options_review_tables
|
|
from lib.options.options_review_images_lib import (
|
|
build_options_review_slot_filename,
|
|
is_valid_options_review_file,
|
|
options_review_upload_dir,
|
|
save_options_review_slot_file,
|
|
)
|
|
from lib.options.options_review_lib import (
|
|
compute_review_stats,
|
|
list_review_trades,
|
|
save_review_entry,
|
|
sync_hedge_plans_closed,
|
|
sync_options_from_exchange,
|
|
upsert_option_history_row,
|
|
)
|
|
|
|
|
|
def _conn() -> sqlite3.Connection:
|
|
c = sqlite3.connect(":memory:")
|
|
c.row_factory = sqlite3.Row
|
|
init_options_review_tables(c)
|
|
init_hedge_plan_tables(c)
|
|
return c
|
|
|
|
|
|
class _FakeFile:
|
|
def __init__(self, name: str, data: bytes = b"img"):
|
|
self.filename = name
|
|
self._data = data
|
|
|
|
def save(self, path: str) -> None:
|
|
Path(path).write_bytes(self._data)
|
|
|
|
|
|
class OptionsReviewTests(unittest.TestCase):
|
|
def test_option_upsert_idempotent(self):
|
|
conn = _conn()
|
|
row = {
|
|
"history_key": "ex:pos1",
|
|
"pos_id": "pos1",
|
|
"inst_id": "ETH-USD-260328-2000-C",
|
|
"underlying": "ETH",
|
|
"opt_type": "C",
|
|
"strike": 2000,
|
|
"sheets": 10,
|
|
"open_avg_px": 0.01,
|
|
"close_avg_px": 0.02,
|
|
"premium_paid": 1.0,
|
|
"realized_pnl": 5.5,
|
|
"created_at": "2026-03-01 10:00:00",
|
|
"closed_at": "2026-03-01 12:00:00",
|
|
"status_label": "已平",
|
|
}
|
|
self.assertEqual(upsert_option_history_row(conn, row), "inserted")
|
|
row["realized_pnl"] = 6.0
|
|
self.assertEqual(upsert_option_history_row(conn, row), "updated")
|
|
n = conn.execute("SELECT COUNT(*) AS c FROM options_review_trades").fetchone()["c"]
|
|
self.assertEqual(n, 1)
|
|
pnl = conn.execute(
|
|
"SELECT realized_pnl_total FROM options_review_trades WHERE history_key='ex:pos1'"
|
|
).fetchone()["realized_pnl_total"]
|
|
self.assertEqual(float(pnl), 6.0)
|
|
|
|
def test_entry_not_overwritten_by_resync(self):
|
|
conn = _conn()
|
|
upsert_option_history_row(
|
|
conn,
|
|
{
|
|
"history_key": "ex:p2",
|
|
"inst_id": "ETH-USD-260328-1800-P",
|
|
"underlying": "ETH",
|
|
"opt_type": "P",
|
|
"realized_pnl": 1.0,
|
|
"created_at": "2026-03-02 10:00:00",
|
|
"closed_at": "2026-03-02 11:00:00",
|
|
},
|
|
)
|
|
tid = conn.execute("SELECT id FROM options_review_trades").fetchone()["id"]
|
|
save_review_entry(
|
|
conn,
|
|
tid,
|
|
{"strategy_tag": "突破追涨", "note": "keep-me", "images": []},
|
|
)
|
|
upsert_option_history_row(
|
|
conn,
|
|
{
|
|
"history_key": "ex:p2",
|
|
"inst_id": "ETH-USD-260328-1800-P",
|
|
"underlying": "ETH",
|
|
"opt_type": "P",
|
|
"realized_pnl": 2.0,
|
|
"created_at": "2026-03-02 10:00:00",
|
|
"closed_at": "2026-03-02 11:00:00",
|
|
},
|
|
)
|
|
note = conn.execute(
|
|
"SELECT note, strategy_tag FROM options_review_entries WHERE trade_id=?",
|
|
(tid,),
|
|
).fetchone()
|
|
self.assertEqual(note["note"], "keep-me")
|
|
self.assertEqual(note["strategy_tag"], "突破追涨")
|
|
pnl = conn.execute(
|
|
"SELECT realized_pnl_total FROM options_review_trades WHERE id=?", (tid,)
|
|
).fetchone()["realized_pnl_total"]
|
|
self.assertEqual(float(pnl), 2.0)
|
|
|
|
def test_hedge_import_and_double_count_guard(self):
|
|
conn = _conn()
|
|
upsert_option_history_row(
|
|
conn,
|
|
{
|
|
"history_key": "ex:leg1",
|
|
"inst_id": "ETH-USD-260328-2000-C",
|
|
"underlying": "ETH",
|
|
"opt_type": "C",
|
|
"realized_pnl": -3.0,
|
|
"created_at": "2026-03-03 09:00:00",
|
|
"closed_at": "2026-03-03 18:00:00",
|
|
},
|
|
)
|
|
plan_id = insert_plan(
|
|
conn,
|
|
{
|
|
"plan_type": SOURCE_PERP_OPTIONS,
|
|
"status": "closed",
|
|
"underlying": "ETH",
|
|
"direction": "long",
|
|
"realized_pnl_perp": 20.0,
|
|
"realized_pnl_options": -3.0,
|
|
"realized_pnl_total": 17.0,
|
|
"close_reason": "tp",
|
|
"opened_at": "2026-03-03 09:00:00",
|
|
"closed_at": "2026-03-03 18:00:00",
|
|
"premium_total": 3.0,
|
|
},
|
|
)
|
|
insert_leg(
|
|
conn,
|
|
{
|
|
"plan_id": plan_id,
|
|
"leg_role": "perp",
|
|
"symbol": "ETH-USDT-SWAP",
|
|
"status": "closed",
|
|
"realized_pnl": 20.0,
|
|
},
|
|
)
|
|
insert_leg(
|
|
conn,
|
|
{
|
|
"plan_id": plan_id,
|
|
"leg_role": "option_hedge",
|
|
"inst_id": "ETH-USD-260328-2000-C",
|
|
"opt_type": "C",
|
|
"status": "closed",
|
|
"realized_pnl": -3.0,
|
|
},
|
|
)
|
|
out = sync_hedge_plans_closed(conn)
|
|
self.assertTrue(out["ok"])
|
|
self.assertEqual(out["inserted"], 1)
|
|
|
|
listed = list_review_trades(conn, include_hedge_legs=False)
|
|
types = {r["source_type"] for r in listed}
|
|
self.assertIn(SOURCE_PERP_OPTIONS, types)
|
|
self.assertNotIn(SOURCE_OPTION, types)
|
|
|
|
listed_all = list_review_trades(conn, include_hedge_legs=True)
|
|
self.assertEqual(len(listed_all), 2)
|
|
|
|
stats = compute_review_stats(conn, include_hedge_legs=False)
|
|
self.assertEqual(stats["kpi"]["total"], 1)
|
|
self.assertEqual(stats["kpi"]["pnl_sum"], 17.0)
|
|
|
|
def test_sync_options_from_mock_exchange(self):
|
|
conn = _conn()
|
|
|
|
def fetch(_ex, limit=500):
|
|
return [
|
|
{
|
|
"instId": "ETH-USD-260328-2100-C",
|
|
"posId": "mock1",
|
|
"openAvgPx": "0.01",
|
|
"closeAvgPx": "0.02",
|
|
"closeTotalPos": "5",
|
|
"realizedPnl": "1.23",
|
|
"type": "2",
|
|
"cTime": "1700000000000",
|
|
"uTime": "1700003600000",
|
|
"uly": "ETH-USD",
|
|
}
|
|
]
|
|
|
|
def fmt(raw, tick_sz=None, ct_mult=0.01):
|
|
return {
|
|
"history_key": f"ex:{raw['posId']}",
|
|
"pos_id": raw["posId"],
|
|
"inst_id": raw["instId"],
|
|
"underlying": "ETH",
|
|
"opt_type": "C",
|
|
"sheets": 5,
|
|
"open_avg_px": 0.01,
|
|
"close_avg_px": 0.02,
|
|
"premium_paid": 0.5,
|
|
"realized_pnl": float(raw["realizedPnl"]),
|
|
"created_at": "2026-01-01 00:00:00",
|
|
"closed_at": "2026-01-01 01:00:00",
|
|
"status_label": "已平",
|
|
}
|
|
|
|
result = sync_options_from_exchange(
|
|
conn, object(), limit=10, fetch_fn=fetch, format_fn=fmt
|
|
)
|
|
self.assertTrue(result["ok"])
|
|
self.assertEqual(result["inserted"], 1)
|
|
row = conn.execute(
|
|
"SELECT realized_pnl_total FROM options_review_trades WHERE history_key='ex:mock1'"
|
|
).fetchone()
|
|
self.assertEqual(float(row["realized_pnl_total"]), 1.23)
|
|
|
|
def test_local_options_trades_import(self):
|
|
conn = _conn()
|
|
from lib.options.options_db import init_options_tables
|
|
|
|
init_options_tables(conn)
|
|
conn.execute(
|
|
"""
|
|
INSERT INTO options_trades
|
|
(inst_id, underlying, opt_type, strike, sheets, eth_amount,
|
|
open_quote, premium_paid, status, realized_pnl, created_at, closed_at)
|
|
VALUES ('ETH-USD-260328-2000-C','ETH','C',2000,2,0.02,0.01,0.5,'closed',3.2,
|
|
'2026-03-01 10:00:00','2026-03-01 12:00:00')
|
|
"""
|
|
)
|
|
from lib.options.options_review_lib import sync_options_from_local_trades
|
|
|
|
out = sync_options_from_local_trades(conn)
|
|
self.assertTrue(out["ok"])
|
|
self.assertEqual(out["inserted"], 1)
|
|
row = conn.execute(
|
|
"SELECT history_key, realized_pnl_total, source_type FROM options_review_trades"
|
|
).fetchone()
|
|
self.assertTrue(str(row["history_key"]).startswith("local_opt:"))
|
|
self.assertEqual(float(row["realized_pnl_total"]), 3.2)
|
|
self.assertEqual(row["source_type"], SOURCE_OPTION)
|
|
|
|
def test_image_namespace(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
folder = options_review_upload_dir(tmp)
|
|
fname = build_options_review_slot_filename(
|
|
"a" * 32, "5m", ".png", secure_filename_fn=lambda x: x
|
|
)
|
|
self.assertTrue(fname.startswith("options_journal_"))
|
|
self.assertTrue(is_valid_options_review_file(fname, "a" * 32, "5m"))
|
|
item = save_options_review_slot_file(
|
|
_FakeFile("x.png"),
|
|
"a" * 32,
|
|
"5m",
|
|
folder,
|
|
secure_filename_fn=lambda x: x,
|
|
)
|
|
self.assertIsNotNone(item)
|
|
self.assertTrue((Path(folder) / item["file"]).is_file())
|
|
|
|
def test_strategy_stats_only_tagged(self):
|
|
conn = _conn()
|
|
upsert_option_history_row(
|
|
conn,
|
|
{
|
|
"history_key": "ex:a",
|
|
"inst_id": "ETH-USD-1-C",
|
|
"underlying": "ETH",
|
|
"opt_type": "C",
|
|
"realized_pnl": 10,
|
|
"created_at": "2026-01-01 00:00:00",
|
|
"closed_at": "2026-01-01 02:00:00",
|
|
},
|
|
)
|
|
upsert_option_history_row(
|
|
conn,
|
|
{
|
|
"history_key": "ex:b",
|
|
"inst_id": "ETH-USD-2-P",
|
|
"underlying": "ETH",
|
|
"opt_type": "P",
|
|
"realized_pnl": -4,
|
|
"created_at": "2026-01-01 00:00:00",
|
|
"closed_at": "2026-01-01 05:00:00",
|
|
},
|
|
)
|
|
tid = conn.execute(
|
|
"SELECT id FROM options_review_trades WHERE history_key='ex:a'"
|
|
).fetchone()["id"]
|
|
save_review_entry(conn, tid, {"strategy_tag": "假破", "images": []})
|
|
stats = compute_review_stats(conn)
|
|
self.assertEqual(len(stats["by_strategy"]), 1)
|
|
self.assertEqual(stats["by_strategy"][0]["key"], "假破")
|
|
self.assertEqual(stats["kpi"]["total"], 2)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|