dekun
53863559f4
Retarget git remote, install path, and deploy docs from crypto_monitor to crypto_monitor_user. Co-authored-by: Cursor <cursoragent@cursor.com>
87 lines
3.9 KiB
Python
87 lines
3.9 KiB
Python
"""期权统计单测."""
|
|
import sqlite3
|
|
from datetime import datetime, timedelta
|
|
from unittest import TestCase
|
|
|
|
from lib.options.options_db import init_options_tables
|
|
from lib.options.options_stats_lib import compute_options_stats, compute_options_stats_from_history
|
|
|
|
|
|
class OptionsStatsLibTests(TestCase):
|
|
def _conn(self):
|
|
conn = sqlite3.connect(":memory:")
|
|
conn.row_factory = sqlite3.Row
|
|
init_options_tables(conn)
|
|
return conn
|
|
|
|
def test_compute_options_stats_empty(self):
|
|
conn = self._conn()
|
|
out = compute_options_stats(lambda: conn)
|
|
self.assertEqual(out["total_closed"], 0)
|
|
self.assertEqual(out["open_count"], 0)
|
|
self.assertIsNone(out["avg_hold_sec"])
|
|
|
|
def test_compute_options_stats_hold_times(self):
|
|
conn = self._conn()
|
|
now = datetime.now()
|
|
win_open = (now - timedelta(hours=2)).strftime("%Y-%m-%d %H:%M:%S")
|
|
win_close = (now - timedelta(hours=1)).strftime("%Y-%m-%d %H:%M:%S")
|
|
loss_open = (now - timedelta(hours=4)).strftime("%Y-%m-%d %H:%M:%S")
|
|
loss_close = (now - timedelta(hours=1)).strftime("%Y-%m-%d %H:%M:%S")
|
|
open_at = (now - timedelta(minutes=30)).strftime("%Y-%m-%d %H:%M:%S")
|
|
conn.execute(
|
|
"""
|
|
INSERT INTO options_trades
|
|
(inst_id, underlying, opt_type, strike, sheets, eth_amount, status,
|
|
realized_pnl, created_at, closed_at)
|
|
VALUES ('A', 'ETH', 'C', 1800, 1, 0.01, 'closed', 1.2, ?, ?)
|
|
""",
|
|
(win_open, win_close),
|
|
)
|
|
conn.execute(
|
|
"""
|
|
INSERT INTO options_trades
|
|
(inst_id, underlying, opt_type, strike, sheets, eth_amount, status,
|
|
realized_pnl, created_at, closed_at)
|
|
VALUES ('B', 'ETH', 'P', 1700, 1, 0.01, 'closed', -1.0, ?, ?)
|
|
""",
|
|
(loss_open, loss_close),
|
|
)
|
|
conn.execute(
|
|
"""
|
|
INSERT INTO options_trades
|
|
(inst_id, underlying, opt_type, strike, sheets, eth_amount, status, created_at)
|
|
VALUES ('C', 'BTC', 'C', 62000, 1, 0.01, 'open', ?)
|
|
""",
|
|
(open_at,),
|
|
)
|
|
conn.commit()
|
|
out = compute_options_stats(lambda: conn)
|
|
self.assertEqual(out["total_closed"], 2)
|
|
self.assertEqual(out["win_count"], 1)
|
|
self.assertEqual(out["loss_count"], 1)
|
|
self.assertEqual(out["win_rate"], 50.0)
|
|
self.assertAlmostEqual(out["avg_win"], 1.2, places=4)
|
|
self.assertAlmostEqual(out["avg_loss"], 1.0, places=4)
|
|
self.assertAlmostEqual(out["avg_win_hold_sec"], 3600.0, delta=5.0)
|
|
self.assertAlmostEqual(out["avg_loss_hold_sec"], 3 * 3600.0, delta=5.0)
|
|
self.assertEqual(out["open_count"], 1)
|
|
self.assertGreater(out["avg_open_hold_sec"], 1700.0)
|
|
|
|
def test_compute_options_stats_from_history_exchange_rows(self):
|
|
history = [
|
|
{"status": "open", "created_at": "2026-07-11 08:08:38"},
|
|
{"status": "closed", "realized_pnl": -3.99, "created_at": "2026-07-09 14:11:46", "closed_at": "2026-07-10 16:00:35"},
|
|
{"status": "closed", "realized_pnl": 0.87, "created_at": "2026-07-09 14:11:46", "closed_at": "2026-07-10 09:55:34"},
|
|
{"status": "closed", "realized_pnl": -1.33, "created_at": "2026-07-08 02:32:44", "closed_at": "2026-07-09 16:00:26"},
|
|
]
|
|
out = compute_options_stats_from_history(history)
|
|
self.assertEqual(out["total_closed"], 3)
|
|
self.assertEqual(out["win_count"], 1)
|
|
self.assertEqual(out["loss_count"], 2)
|
|
self.assertAlmostEqual(out["avg_win"], 0.87, places=4)
|
|
self.assertAlmostEqual(out["avg_loss"], 2.66, places=2)
|
|
self.assertAlmostEqual(out["profit_loss_ratio"], 0.33, places=2)
|
|
self.assertEqual(out["open_count"], 1)
|
|
self.assertAlmostEqual(out["net_realized_pnl"], round(0.87 - 3.99 - 1.33, 4), places=4)
|