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()