151 lines
5.2 KiB
Python
151 lines
5.2 KiB
Python
from fastapi.testclient import TestClient
|
|
import json
|
|
from datetime import datetime
|
|
|
|
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
|
|
|
|
|
|
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"] == "前瞻信号不足"
|