120 lines
4.1 KiB
Python
120 lines
4.1 KiB
Python
import os
|
|
import sys
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
|
|
PROJECT_DIR = os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))
|
|
if PROJECT_DIR not in sys.path:
|
|
sys.path.insert(0, PROJECT_DIR)
|
|
|
|
from app.db import altcoin_db
|
|
from app.web import web_server
|
|
from test_actionable_active_recommendations import _insert_recommendation
|
|
|
|
|
|
@pytest.fixture
|
|
def temp_db(monkeypatch, tmp_path):
|
|
db_path = tmp_path / "altcoin_monitor.db"
|
|
monkeypatch.setattr(altcoin_db, "DB_PATH", str(db_path))
|
|
monkeypatch.setattr(web_server, "get_stats", altcoin_db.get_stats)
|
|
monkeypatch.setattr(web_server, "get_active_recommendations", altcoin_db.get_active_recommendations)
|
|
monkeypatch.setattr(web_server, "get_active_recommendations_deduped", altcoin_db.get_active_recommendations_deduped)
|
|
monkeypatch.setattr(web_server, "get_all_recommendations", altcoin_db.get_all_recommendations)
|
|
altcoin_db.init_db()
|
|
return db_path
|
|
|
|
|
|
def test_actionable_plans_are_not_counted_as_executed_pnl_until_entry_triggers(temp_db):
|
|
_insert_recommendation(
|
|
temp_db,
|
|
symbol='BUY/USDT',
|
|
action_status='可即刻买入',
|
|
entry_plan_json='{"entry_action": "可即刻买入"}',
|
|
pnl_pct=5.0,
|
|
max_pnl_pct=5.0,
|
|
)
|
|
_insert_recommendation(
|
|
temp_db,
|
|
symbol='WAIT/USDT',
|
|
action_status='等回踩',
|
|
entry_plan_json='{"entry_action": "等回踩", "entry_price": 90}',
|
|
pnl_pct=12.0,
|
|
max_pnl_pct=12.0,
|
|
)
|
|
_insert_recommendation(
|
|
temp_db,
|
|
symbol='OBS/USDT',
|
|
action_status='持有',
|
|
entry_plan_json='{"entry_action": "继续观察"}',
|
|
pnl_pct=20.0,
|
|
max_pnl_pct=20.0,
|
|
)
|
|
|
|
stats = altcoin_db.get_stats()
|
|
live = stats['live_overview']
|
|
|
|
# 现可买 + 等回踩仍属于实时可操作看板数量
|
|
assert live['actionable_count'] == 2
|
|
assert live['buy_now_count'] == 1
|
|
assert live['wait_pullback_count'] == 1
|
|
|
|
# 但 buy_now 只是入场窗口，不等于已成交；未触发入场前不计算收益。
|
|
assert live['executed_trade_count'] == 0
|
|
assert live['executed_pnl_sum'] == pytest.approx(0.0)
|
|
assert live['executed_avg_pnl'] == pytest.approx(0.0)
|
|
assert stats['active_pnl_sum'] == pytest.approx(0.0)
|
|
assert stats['active_avg_pnl'] == pytest.approx(0.0)
|
|
assert stats['active_success_count'] == 0
|
|
assert stats['active_pending_count'] == 0
|
|
|
|
|
|
def test_entry_triggered_buy_now_counts_as_executed_pnl(temp_db):
|
|
_insert_recommendation(
|
|
temp_db,
|
|
symbol='BUY/USDT',
|
|
action_status='可即刻买入',
|
|
entry_plan_json='{"entry_action": "可即刻买入"}',
|
|
pnl_pct=5.0,
|
|
max_pnl_pct=5.0,
|
|
)
|
|
conn = altcoin_db.sqlite3.connect(str(temp_db))
|
|
conn.execute("UPDATE recommendation SET entry_triggered=1 WHERE symbol='BUY/USDT'")
|
|
conn.commit()
|
|
conn.close()
|
|
|
|
stats = altcoin_db.get_stats()
|
|
live = stats['live_overview']
|
|
assert live['executed_trade_count'] == 1
|
|
assert live['executed_pnl_sum'] == pytest.approx(5.0)
|
|
|
|
|
|
def test_active_api_marks_wait_plan_as_unexecuted_not_success(temp_db):
|
|
_insert_recommendation(
|
|
temp_db,
|
|
symbol='WAIT/USDT',
|
|
action_status='等回踩',
|
|
entry_plan_json='{"entry_action": "等回踩", "entry_price": 90}',
|
|
pnl_pct=12.0,
|
|
max_pnl_pct=12.0,
|
|
)
|
|
|
|
client = TestClient(web_server.app)
|
|
rows = client.get('/api/recommendations/active').json()
|
|
assert len(rows) == 1
|
|
assert rows[0]['execution_status'] == 'wait_pullback'
|
|
assert rows[0]['recommendation_result'] == 'pending'
|
|
assert rows[0]['recommendation_result_label'] == '⏳ 未执行'
|
|
|
|
|
|
def test_dashboard_kline_only_marks_executed_entry_price(temp_db, monkeypatch):
|
|
monkeypatch.setattr(web_server, "_require_active_subscription", lambda altcoin_session='': ({"id": 1}, {"plan_code": "test"}))
|
|
client = TestClient(web_server.app)
|
|
html = client.get('/app').text
|
|
# v1.7.7: 新看板使用 drawPin 函数 + pin标记
|
|
assert 'drawPin' in html
|
|
assert 'data-entry-price=' in html
|
|
assert '止损 ' in html
|
|
assert '止盈 ' in html
|
|
assert '推荐 ${recText}' not in html
|