216 lines
8.3 KiB
Python
216 lines
8.3 KiB
Python
from fastapi.testclient import TestClient
|
|
import json
|
|
from datetime import datetime, timedelta
|
|
|
|
from app.db import auth_db
|
|
from app.db.review_center import get_review_center_dashboard
|
|
from app.web import web_server
|
|
|
|
|
|
def _login_user(email: str, admin: bool = False) -> str:
|
|
reg = auth_db.register_user(email, "StrongPass123")
|
|
auth_db.verify_email(email, reg["verification_code"])
|
|
user = auth_db.get_user_by_email(email)
|
|
auth_db.claim_free_trial(user["id"])
|
|
if admin:
|
|
auth_db.set_user_admin(email, True)
|
|
return auth_db.login_user(email, "StrongPass123")["token"]
|
|
|
|
|
|
def test_review_center_page_and_api_require_admin():
|
|
token = _login_user("normal-review-center@example.com")
|
|
client = TestClient(web_server.app)
|
|
client.cookies.set("altcoin_session", token)
|
|
|
|
page = client.get("/review-center")
|
|
api = client.get("/api/review-center/dashboard")
|
|
|
|
assert page.status_code == 403
|
|
assert api.status_code == 403
|
|
|
|
|
|
def test_review_center_admin_can_access_page_and_api():
|
|
token = _login_user("admin-review-center@example.com", admin=True)
|
|
client = TestClient(web_server.app)
|
|
client.cookies.set("altcoin_session", token)
|
|
|
|
page = client.get("/review-center")
|
|
api = client.get("/api/review-center/dashboard")
|
|
|
|
assert page.status_code == 200
|
|
assert "复盘中心" in page.text
|
|
assert api.status_code == 200
|
|
data = api.json()
|
|
assert "opportunity" in data
|
|
assert "paper_trading" in data
|
|
assert "evidence" in data
|
|
assert "iteration" in data
|
|
|
|
|
|
def test_review_center_separates_opportunity_and_paper_pnl(pg_conn):
|
|
pg_conn.execute(
|
|
"""
|
|
INSERT INTO recommendation (
|
|
symbol, rec_time, rec_state, rec_score, entry_price,
|
|
status, execution_status, action_status, display_bucket, entry_triggered
|
|
) VALUES
|
|
('OBS/USDT', '2026-05-16T10:00:00', '观察', 10, 1, 'active', 'observe', '观察', 'watch_pool', 0),
|
|
('BUY/USDT', '2026-05-16T11:00:00', '爆发', 30, 10, 'active', 'buy_now', '可即刻买入', 'realtime', 1)
|
|
"""
|
|
)
|
|
pg_conn.execute(
|
|
"""
|
|
INSERT INTO paper_trades (
|
|
recommendation_id, symbol, side, status, opened_at, closed_at,
|
|
entry_price, exit_price, qty, notional_usdt, margin_usdt, leverage,
|
|
current_price, pnl_pct, realized_pnl_pct, realized_pnl_usdt,
|
|
source_status, source_action, created_at, updated_at
|
|
) VALUES (
|
|
2, 'BUY/USDT', 'long', 'closed', '2026-05-16T11:01:00', '2026-05-16T12:00:00',
|
|
10, 11, 500, 5000, 1000, 5,
|
|
11, 10, 10, 490,
|
|
'buy_now', '可即刻买入', '2026-05-16T11:01:00', '2026-05-16T12:00:00'
|
|
)
|
|
"""
|
|
)
|
|
pg_conn.commit()
|
|
|
|
data = get_review_center_dashboard(days=30)
|
|
|
|
assert data["opportunity"]["summary"]["total_opportunities"] == 2
|
|
assert data["opportunity"]["summary"]["paper_executed_count"] == 1
|
|
assert "total_pnl_usdt" not in data["opportunity"]["summary"]
|
|
assert data["paper_trading"]["summary"]["closed_count"] == 1
|
|
assert data["paper_trading"]["summary"]["realized_pnl_usdt"] == 490
|
|
assert data["paper_trading"]["trade_attribution"]["entry_path"][0]["closed_trade_count"] == 1
|
|
assert "watch_order_attribution" in data["paper_trading"]
|
|
|
|
|
|
def test_review_center_strategy_counts_use_same_closed_trade_window(pg_conn):
|
|
now = datetime.now()
|
|
recent_close = now - timedelta(days=2)
|
|
old_close = now - timedelta(days=40)
|
|
old_open_recent_close = now - timedelta(days=20)
|
|
outside_open = now - timedelta(days=60)
|
|
strategy_code = "volume_ignition_1h_v1"
|
|
|
|
pg_conn.execute(
|
|
"""
|
|
INSERT INTO recommendation (
|
|
id, symbol, rec_time, rec_state, rec_score, entry_price,
|
|
status, execution_status, action_status, display_bucket, entry_triggered,
|
|
strategy_code, strategy_version, entry_plan_json
|
|
) VALUES
|
|
(101, 'WIN/USDT', %s, '爆发', 30, 10, 'active', 'buy_now', '可即刻买入', 'realtime', 1, %s, 'v-test', %s),
|
|
(102, 'OLD/USDT', %s, '爆发', 30, 10, 'active', 'buy_now', '可即刻买入', 'realtime', 1, %s, 'v-test', %s)
|
|
""",
|
|
(
|
|
old_open_recent_close.isoformat(timespec="seconds"),
|
|
strategy_code,
|
|
json.dumps({"strategy_code": strategy_code}, ensure_ascii=False),
|
|
outside_open.isoformat(timespec="seconds"),
|
|
strategy_code,
|
|
json.dumps({"strategy_code": strategy_code}, ensure_ascii=False),
|
|
),
|
|
)
|
|
pg_conn.execute(
|
|
"""
|
|
INSERT INTO paper_trades (
|
|
recommendation_id, symbol, side, status, opened_at, closed_at,
|
|
entry_price, exit_price, qty, notional_usdt, margin_usdt, leverage,
|
|
current_price, pnl_pct, realized_pnl_pct, realized_pnl_usdt,
|
|
source_status, source_action, strategy_code, created_at, updated_at
|
|
) VALUES
|
|
(101, 'WIN/USDT', 'long', 'closed', %s, %s, 10, 11, 500, 5000, 1000, 5, 11, 10, 10, 500, 'buy_now', '可即刻买入', %s, %s, %s),
|
|
(102, 'OLD/USDT', 'long', 'closed', %s, %s, 10, 9, 500, 5000, 1000, 5, 9, -10, -10, -500, 'buy_now', '可即刻买入', %s, %s, %s)
|
|
""",
|
|
(
|
|
old_open_recent_close.isoformat(timespec="seconds"),
|
|
recent_close.isoformat(timespec="seconds"),
|
|
strategy_code,
|
|
old_open_recent_close.isoformat(timespec="seconds"),
|
|
recent_close.isoformat(timespec="seconds"),
|
|
outside_open.isoformat(timespec="seconds"),
|
|
old_close.isoformat(timespec="seconds"),
|
|
strategy_code,
|
|
outside_open.isoformat(timespec="seconds"),
|
|
old_close.isoformat(timespec="seconds"),
|
|
),
|
|
)
|
|
pg_conn.commit()
|
|
|
|
data = get_review_center_dashboard(days=30)
|
|
board = next(x for x in data["strategy_evaluation"]["strategies"] if x["strategy_code"] == strategy_code)
|
|
lower = next(x for x in data["paper_trading"]["trade_attribution"]["strategy_code"] if x["strategy_code"] == strategy_code)
|
|
|
|
assert board["closed_trade_count"] == 1
|
|
assert lower["closed_trade_count"] == 1
|
|
assert data["paper_trading"]["summary"]["closed_count"] == 1
|
|
assert board["realized_pnl_usdt"] == 500
|
|
assert lower["realized_pnl_usdt"] == 500
|
|
|
|
|
|
def test_review_center_iteration_digest_summarizes_actions(pg_conn):
|
|
now = datetime.now().isoformat(timespec="seconds")
|
|
today = now[:10]
|
|
changed_rules = [
|
|
{
|
|
"type": "factor_weight_governance",
|
|
"signal": "breakout_pullback_d1",
|
|
"action": "升权",
|
|
"old_weight": 1.0,
|
|
"new_weight": 1.8,
|
|
"sample_size": 12,
|
|
"hit_rate": 70,
|
|
"avg_pnl": 4.2,
|
|
},
|
|
{
|
|
"type": "factor_weight_governance",
|
|
"signal": "unknown",
|
|
"action": "降权",
|
|
"old_weight": 0.3,
|
|
"new_weight": 0.15,
|
|
"sample_size": 13,
|
|
"hit_rate": 15.4,
|
|
"avg_pnl": 0.9,
|
|
},
|
|
]
|
|
pg_conn.execute(
|
|
"""
|
|
INSERT INTO strategy_iteration_log (
|
|
run_date, created_at, title, changed_rules_json, metrics_json,
|
|
release_decision, release_reason, strategy_version
|
|
) VALUES (
|
|
%s, %s, '第1轮复盘迭代',
|
|
%s, %s, 'hold', '继续观察', 'v1.7.11'
|
|
)
|
|
""",
|
|
(
|
|
today,
|
|
now,
|
|
json.dumps(changed_rules, ensure_ascii=False),
|
|
json.dumps({"factor_weight_updates": 2, "effective_review_count": 3}, ensure_ascii=False),
|
|
),
|
|
)
|
|
pg_conn.execute(
|
|
"""
|
|
INSERT INTO strategy_rule_candidate (
|
|
created_at, source, rule_type, signal_name, rule_description,
|
|
sample_size, confidence_score, status
|
|
) VALUES (
|
|
%s, 'dual_attribution_failure', 'penalty',
|
|
'前瞻信号不足', '失败模式进入灰度', 10, 95, 'gray'
|
|
)
|
|
""",
|
|
(now,),
|
|
)
|
|
pg_conn.commit()
|
|
|
|
data = get_review_center_dashboard(days=30)
|
|
digest = data["iteration"]["digest"]
|
|
|
|
assert digest["latest"]["metrics"]["factor_weight_updates"] == 2
|
|
assert digest["upgraded"][0]["signal"] == "breakout_pullback_d1"
|
|
assert digest["downgraded"][0]["signal"] == "unknown"
|
|
assert digest["gray"][0]["signal"] == "前瞻信号不足"
|