dekun/crypto_monitor_user
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Initialize crypto_monitor_user (user edition) from monitor codebase.

Browse Source 
Retarget git remote, install path, and deploy docs from crypto_monitor to crypto_monitor_user.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
dekun
2026-07-17 16:18:13 +08:00
commit 53863559f4
608 changed files with 162764 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_options_review_lib.py
+342
View File
@@ -0,0 +1,342 @@
"""期权复盘(含对冲)单元测试:导入去重、双计防护、复盘不被覆盖、统计."""
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_hide_trade_persists_across_local_sync(self):
conn = _conn()
from lib.options.options_db import init_options_tables
from lib.options.options_review_lib import (
hide_review_trade,
sync_options_from_local_trades,
)
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-1-C','ETH','C',2000,1,0.01,0.01,0.2,'closed',1.0,
'2026-03-01 10:00:00','2026-03-01 11:00:00')
"""
)
sync_options_from_local_trades(conn)
tid = conn.execute("SELECT id FROM options_review_trades").fetchone()["id"]
out = hide_review_trade(conn, tid)
self.assertTrue(out["ok"])
self.assertEqual(
conn.execute("SELECT COUNT(*) AS c FROM options_review_trades").fetchone()["c"],
0,
)
sync_options_from_local_trades(conn)
self.assertEqual(
conn.execute("SELECT COUNT(*) AS c FROM options_review_trades").fetchone()["c"],
0,
)
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()