dekun
3052607280
Include strategy_trade_snapshots when trade_records is empty, harden SQL for older schemas, and show per-exchange sync errors in the archive UI. Co-authored-by: Cursor <cursoragent@cursor.com>
103 lines
3.4 KiB
Python
103 lines
3.4 KiB
Python
"""档案交易：strategy_trade_snapshots 补全 gate_bot 漏记。"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import sqlite3
|
|
import tempfile
|
|
from datetime import datetime, timedelta
|
|
from pathlib import Path
|
|
|
|
from hub_trades_lib import fetch_trades_for_archive
|
|
|
|
|
|
def _init_db(path: Path) -> sqlite3.Connection:
|
|
conn = sqlite3.connect(str(path))
|
|
conn.row_factory = sqlite3.Row
|
|
conn.execute(
|
|
"""
|
|
CREATE TABLE trade_records (
|
|
id INTEGER PRIMARY KEY,
|
|
symbol TEXT,
|
|
direction TEXT,
|
|
result TEXT,
|
|
pnl_amount REAL,
|
|
opened_at TEXT,
|
|
closed_at TEXT,
|
|
opened_at_ms INTEGER,
|
|
closed_at_ms INTEGER,
|
|
created_at TEXT,
|
|
trend_plan_id INTEGER
|
|
)
|
|
"""
|
|
)
|
|
conn.execute(
|
|
"""
|
|
CREATE TABLE strategy_trade_snapshots (
|
|
id INTEGER PRIMARY KEY,
|
|
strategy_type TEXT,
|
|
source_id INTEGER,
|
|
symbol TEXT,
|
|
direction TEXT,
|
|
result_label TEXT,
|
|
status_at_close TEXT,
|
|
opened_at TEXT,
|
|
closed_at TEXT,
|
|
pnl_amount REAL,
|
|
snapshot_json TEXT,
|
|
created_at TEXT
|
|
)
|
|
"""
|
|
)
|
|
return conn
|
|
|
|
|
|
def test_merge_snapshot_when_trade_record_missing():
|
|
with tempfile.TemporaryDirectory() as td:
|
|
conn = _init_db(Path(td) / "t.db")
|
|
closed = (datetime.now() - timedelta(days=1)).strftime("%Y-%m-%d %H:%M:%S")
|
|
conn.execute(
|
|
"""
|
|
INSERT INTO strategy_trade_snapshots (
|
|
id, strategy_type, source_id, symbol, direction,
|
|
result_label, opened_at, closed_at, pnl_amount, snapshot_json, created_at
|
|
) VALUES (?,?,?,?,?,?,?,?,?,?,?)
|
|
""",
|
|
(7, "trend_pullback", 42, "ONDO/USDT", "long", "止损", closed, closed, -1.2, "{}", closed),
|
|
)
|
|
conn.commit()
|
|
trades = fetch_trades_for_archive(conn, days=30, limit=50)
|
|
conn.close()
|
|
assert len(trades) == 1
|
|
assert trades[0]["symbol"] == "ONDO/USDT"
|
|
assert trades[0]["id"] == -7
|
|
assert trades[0].get("from_snapshot") is True
|
|
|
|
|
|
def test_skip_snapshot_when_trade_record_exists():
|
|
with tempfile.TemporaryDirectory() as td:
|
|
conn = _init_db(Path(td) / "t.db")
|
|
closed = (datetime.now() - timedelta(days=1)).strftime("%Y-%m-%d %H:%M:%S")
|
|
conn.execute(
|
|
"""
|
|
INSERT INTO trade_records (
|
|
id, symbol, direction, result, pnl_amount,
|
|
opened_at, closed_at, opened_at_ms, closed_at_ms, created_at, trend_plan_id
|
|
) VALUES (?,?,?,?,?,?,?,?,?,?,?)
|
|
""",
|
|
(1, "ONDO/USDT", "long", "止损", -1.2, closed, closed, 1, 2, closed, 42),
|
|
)
|
|
conn.execute(
|
|
"""
|
|
INSERT INTO strategy_trade_snapshots (
|
|
id, strategy_type, source_id, symbol, direction,
|
|
result_label, opened_at, closed_at, pnl_amount, snapshot_json, created_at
|
|
) VALUES (?,?,?,?,?,?,?,?,?,?,?)
|
|
""",
|
|
(7, "trend_pullback", 42, "ONDO/USDT", "long", "止损", closed, closed, -1.2, "{}", closed),
|
|
)
|
|
conn.commit()
|
|
trades = fetch_trades_for_archive(conn, days=30, limit=50)
|
|
conn.close()
|
|
assert len(trades) == 1
|
|
assert trades[0]["id"] == 1
|