226 lines
9.1 KiB
Python
226 lines
9.1 KiB
Python
import json
|
|
import os
|
|
import sqlite3
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from datetime import datetime
|
|
from unittest.mock import patch
|
|
|
|
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
|
|
|
|
|
|
class RecommendationHistoryBase(unittest.TestCase):
|
|
def setUp(self):
|
|
self.tmpdir = tempfile.TemporaryDirectory()
|
|
self.db_path = os.path.join(self.tmpdir.name, 'test_altcoin.db')
|
|
self.db_patch = patch.object(altcoin_db, 'DB_PATH', self.db_path)
|
|
self.db_patch.start()
|
|
altcoin_db.init_db()
|
|
|
|
def tearDown(self):
|
|
self.db_patch.stop()
|
|
self.tmpdir.cleanup()
|
|
|
|
def _insert_rec(self, **kwargs):
|
|
defaults = dict(
|
|
symbol='AAA/USDT',
|
|
rec_time=datetime.now().strftime('%Y-%m-%dT%H:%M:%S'),
|
|
rec_state='加速',
|
|
rec_score=40,
|
|
entry_price=100.0,
|
|
stop_loss=95.0,
|
|
tp1=110.0,
|
|
tp2=118.0,
|
|
sector='AI',
|
|
signals=json.dumps(['🟢 15min即刻入场信号'], ensure_ascii=False),
|
|
is_meme=0,
|
|
status='active',
|
|
current_price=100.0,
|
|
max_price=104.0,
|
|
min_price=98.0,
|
|
pnl_pct=0.0,
|
|
max_pnl_pct=4.0,
|
|
max_drawdown_pct=-1.0,
|
|
hit_tp1_time='',
|
|
hit_tp2_time='',
|
|
stopped_out_time='',
|
|
expired_time='',
|
|
last_track_time=datetime.now().strftime('%Y-%m-%dT%H:%M:%S'),
|
|
entry_plan_json=json.dumps({
|
|
'entry_price': 100.0,
|
|
'entry_action': '可即刻买入',
|
|
'risk_reward_ok': True,
|
|
'stop_loss': 95.0,
|
|
'stop_pct': -5.0,
|
|
'tp1': 110.0,
|
|
'tp2': 118.0,
|
|
'rr1': 2.0,
|
|
'rr2': 3.6,
|
|
'entry_trigger_confirmed': True,
|
|
}, ensure_ascii=False),
|
|
action_status='可即刻买入',
|
|
direction='多头启动',
|
|
)
|
|
defaults.update(kwargs)
|
|
conn = sqlite3.connect(self.db_path)
|
|
conn.execute(
|
|
'''
|
|
INSERT INTO recommendation (
|
|
symbol, rec_time, rec_state, rec_score, entry_price, stop_loss, tp1, tp2,
|
|
sector, signals, is_meme, status, current_price, max_price, min_price,
|
|
pnl_pct, max_pnl_pct, max_drawdown_pct, hit_tp1_time, hit_tp2_time,
|
|
stopped_out_time, expired_time, last_track_time, entry_plan_json,
|
|
action_status, direction
|
|
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
|
''',
|
|
(
|
|
defaults['symbol'], defaults['rec_time'], defaults['rec_state'], defaults['rec_score'],
|
|
defaults['entry_price'], defaults['stop_loss'], defaults['tp1'], defaults['tp2'],
|
|
defaults['sector'], defaults['signals'], defaults['is_meme'], defaults['status'],
|
|
defaults['current_price'], defaults['max_price'], defaults['min_price'], defaults['pnl_pct'],
|
|
defaults['max_pnl_pct'], defaults['max_drawdown_pct'], defaults['hit_tp1_time'], defaults['hit_tp2_time'],
|
|
defaults['stopped_out_time'], defaults['expired_time'], defaults['last_track_time'], defaults['entry_plan_json'],
|
|
defaults['action_status'], defaults['direction'],
|
|
),
|
|
)
|
|
conn.commit()
|
|
conn.close()
|
|
|
|
|
|
class RecommendationHistoryGroupingTests(RecommendationHistoryBase):
|
|
def test_get_all_recommendations_exposes_each_history_group(self):
|
|
self._insert_rec(symbol='AAA/USDT', action_status='可即刻买入', status='active')
|
|
self._insert_rec(
|
|
symbol='BBB/USDT',
|
|
action_status='等回踩',
|
|
status='active',
|
|
current_price=105.0,
|
|
entry_plan_json=json.dumps({'entry_action': '等回踩', 'entry_price': 100.0}, ensure_ascii=False),
|
|
)
|
|
self._insert_rec(
|
|
symbol='CCC/USDT',
|
|
action_status='持有',
|
|
status='active',
|
|
entry_plan_json=json.dumps({'entry_action': '继续观察', 'entry_price': 100.0}, ensure_ascii=False),
|
|
)
|
|
self._insert_rec(symbol='DDD/USDT', action_status='衰减', status='active')
|
|
self._insert_rec(symbol='EEE/USDT', action_status='止盈1', status='hit_tp1')
|
|
|
|
rows = altcoin_db.get_all_recommendations(limit=20)
|
|
mapping = {row['symbol']: row['execution_status'] for row in rows}
|
|
|
|
self.assertEqual(mapping['AAA/USDT'], 'buy_now')
|
|
self.assertEqual(mapping['BBB/USDT'], 'wait_pullback')
|
|
self.assertEqual(mapping['CCC/USDT'], 'observe')
|
|
self.assertEqual(mapping['DDD/USDT'], 'invalid')
|
|
self.assertEqual(mapping['EEE/USDT'], 'completed')
|
|
|
|
def test_get_all_recommendations_can_drive_history_summary_counts(self):
|
|
self._insert_rec(symbol='AAA/USDT', action_status='可即刻买入', status='active')
|
|
self._insert_rec(symbol='BBB/USDT', action_status='可即刻买入', status='active')
|
|
self._insert_rec(
|
|
symbol='CCC/USDT',
|
|
action_status='等回踩',
|
|
status='active',
|
|
current_price=105.0,
|
|
entry_plan_json=json.dumps({'entry_action': '等回踩', 'entry_price': 100.0}, ensure_ascii=False),
|
|
)
|
|
self._insert_rec(
|
|
symbol='DDD/USDT',
|
|
action_status='持有',
|
|
status='active',
|
|
entry_plan_json=json.dumps({'entry_action': '继续观察', 'entry_price': 100.0}, ensure_ascii=False),
|
|
)
|
|
self._insert_rec(symbol='EEE/USDT', action_status='衰减', status='active')
|
|
self._insert_rec(symbol='FFF/USDT', action_status='止盈2', status='hit_tp2')
|
|
|
|
rows = altcoin_db.get_all_recommendations(limit=20)
|
|
counts = {}
|
|
for row in rows:
|
|
key = row['execution_status']
|
|
counts[key] = counts.get(key, 0) + 1
|
|
|
|
self.assertEqual(counts, {
|
|
'buy_now': 2,
|
|
'wait_pullback': 1,
|
|
'observe': 1,
|
|
'invalid': 1,
|
|
'completed': 1,
|
|
})
|
|
|
|
|
|
class DecisionModeHistoryTests(RecommendationHistoryBase):
|
|
def test_get_all_recommendations_supports_decision_only_history_mode(self):
|
|
self._insert_rec(symbol='BUY/USDT', action_status='可即刻买入', status='active')
|
|
self._insert_rec(
|
|
symbol='WAIT/USDT',
|
|
action_status='等回踩',
|
|
status='active',
|
|
current_price=105.0,
|
|
entry_plan_json=json.dumps({'entry_action': '等回踩', 'entry_price': 100.0}, ensure_ascii=False),
|
|
)
|
|
self._insert_rec(
|
|
symbol='OBS/USDT',
|
|
action_status='持有',
|
|
status='active',
|
|
entry_plan_json=json.dumps({'entry_action': '继续观察', 'entry_price': 100.0}, ensure_ascii=False),
|
|
)
|
|
self._insert_rec(symbol='INV/USDT', action_status='衰减', status='active')
|
|
self._insert_rec(symbol='TP/USDT', action_status='止盈1', status='hit_tp1')
|
|
self._insert_rec(symbol='STOP/USDT', action_status='止损', status='stopped_out')
|
|
|
|
rows = altcoin_db.get_all_recommendations(limit=20, decision_only=True)
|
|
mapping = {row['symbol']: row['execution_status'] for row in rows}
|
|
|
|
self.assertEqual(set(mapping.keys()), {'TP/USDT', 'STOP/USDT'})
|
|
self.assertEqual(mapping['TP/USDT'], 'completed')
|
|
self.assertEqual(mapping['STOP/USDT'], 'invalid')
|
|
|
|
def test_stopped_out_with_prior_float_profit_stays_failed_in_history_summary(self):
|
|
self._insert_rec(
|
|
symbol='MLN/USDT',
|
|
action_status='止损',
|
|
status='stopped_out',
|
|
entry_price=3.61,
|
|
current_price=3.12,
|
|
max_price=3.80,
|
|
min_price=3.12,
|
|
pnl_pct=-13.57,
|
|
max_pnl_pct=5.26,
|
|
max_drawdown_pct=-13.57,
|
|
stopped_out_time='2026-05-14T21:21:07',
|
|
last_track_time='2026-05-14T21:21:07',
|
|
)
|
|
|
|
result, label = altcoin_db._classify_recommendation_result({
|
|
'symbol': 'MLN/USDT',
|
|
'status': 'stopped_out',
|
|
'action_status': '止损',
|
|
'entry_price': 3.61,
|
|
'current_price': 3.12,
|
|
'pnl_pct': -13.57,
|
|
'max_pnl_pct': 5.26,
|
|
'max_drawdown_pct': -13.57,
|
|
})
|
|
self.assertEqual(result, 'failed')
|
|
self.assertIn('止损', label)
|
|
|
|
page = altcoin_db.get_all_recommendations(limit=20, decision_only=True, with_meta=True)
|
|
item = page['items'][0]
|
|
self.assertEqual(item['symbol'], 'MLN/USDT')
|
|
self.assertEqual(item['recommendation_result'], 'failed')
|
|
self.assertEqual(item['execution_status'], 'invalid')
|
|
self.assertEqual(page['summary']['success_count'], 0)
|
|
self.assertEqual(page['summary']['failure_count'], 1)
|
|
self.assertAlmostEqual(page['summary']['total_pnl'], -13.57, places=2)
|
|
self.assertAlmostEqual(page['summary']['best_pnl'], -13.57, places=2)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|