import json import os import sqlite3 import tempfile import unittest from datetime import datetime from unittest.mock import patch from app.db import altcoin_db class RecommendationExecutionStatusTests(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() def _get_active(self): rows = altcoin_db.get_active_recommendations_deduped() self.assertGreaterEqual(len(rows), 1) return next(r for r in rows if r['symbol'] == 'AAA/USDT') def test_buy_now_status_for_immediate_entry(self): self._insert_rec() row = self._get_active() self.assertEqual(row['execution_status'], 'buy_now') self.assertEqual(row['execution_label'], '🟢 现在可买') self.assertEqual(row['initial_action'], '可即刻买入') self.assertIn('推荐时就是可即刻买入', row['execution_reason']) def test_low_score_buy_now_downgrades_before_dashboard(self): self._insert_rec(symbol='DBG/USDT', rec_score=8) rows = altcoin_db.get_active_recommendations_deduped(actionable_only=False) row = next(r for r in rows if r['symbol'] == 'DBG/USDT') self.assertEqual(row['execution_status'], 'observe') self.assertEqual(row['action_status'], '观察') self.assertIn('信号不足', row['execution_reason']) self.assertIn('entry_quality_gate', row['entry_plan']) def test_stale_buy_now_downgrades_before_dashboard(self): self._insert_rec(rec_time='2026-04-29T10:00:00', last_track_time='2026-04-29T10:05:00') rows = altcoin_db.get_active_recommendations_deduped(actionable_only=False) row = next(r for r in rows if r['symbol'] == 'AAA/USDT') self.assertNotEqual(row['execution_status'], 'buy_now') self.assertIn(row['action_status'], ('观察', '等回踩')) self.assertTrue(row.get('entry_window_alert')) def test_wait_pullback_status_for_wait_action(self): self._insert_rec( symbol='BBB/USDT', action_status='等回踩', current_price=104.0, entry_plan_json=json.dumps({ 'entry_price': 100.0, 'entry_action': '等回踩', 'risk_reward_ok': True, }, ensure_ascii=False), ) rows = altcoin_db.get_active_recommendations_deduped() target = next(r for r in rows if r['symbol'] == 'BBB/USDT') self.assertEqual(target['execution_status'], 'wait_pullback') self.assertEqual(target['execution_label'], '🟡 等回踩，不追高') self.assertIn('等待回踩', target['execution_reason']) def test_wait_pullback_downbreak_degrades_to_observe(self): self._insert_rec( symbol='BBB/USDT', action_status='等回踩', current_price=98.0, entry_plan_json=json.dumps({ 'entry_price': 100.0, 'entry_action': '等回踩', 'risk_reward_ok': True, 'stop_loss': 95.0, 'tp1': 110.0, 'rr1': 2.0, }, ensure_ascii=False), ) rows = altcoin_db.get_active_recommendations_deduped(actionable_only=False) target = next(r for r in rows if r['symbol'] == 'BBB/USDT') self.assertEqual(target['action_status'], '观察') self.assertEqual(target['execution_status'], 'observe') self.assertIn('回踩参考已下破', target['execution_reason']) def test_wait_pullback_at_reference_promotes_to_buy_now_when_rr_ok(self): self._insert_rec( symbol='BBB/USDT', action_status='等回踩', current_price=99.9, entry_plan_json=json.dumps({ 'entry_price': 100.0, 'entry_action': '等回踩', 'risk_reward_ok': True, 'stop_loss': 95.0, 'tp1': 110.0, 'rr1': 2.0, }, ensure_ascii=False), ) rows = altcoin_db.get_active_recommendations_deduped(actionable_only=False) target = next(r for r in rows if r['symbol'] == 'BBB/USDT') self.assertEqual(target['action_status'], '可即刻买入') self.assertEqual(target['execution_status'], 'buy_now') self.assertTrue(target['entry_plan']['entry_trigger_confirmed']) def test_invalid_status_for_decay(self): self._insert_rec( symbol='CCC/USDT', action_status='衰减', current_price=107.0, max_pnl_pct=8.0, entry_plan_json=json.dumps({ 'entry_price': 100.0, 'entry_action': '可即刻买入', 'risk_reward_ok': True, }, ensure_ascii=False), ) rows = altcoin_db.get_active_recommendations_deduped(actionable_only=False) target = next(r for r in rows if r['symbol'] == 'CCC/USDT') self.assertEqual(target['execution_status'], 'invalid') self.assertEqual(target['execution_label'], '🔴 已失效，勿追') self.assertIn('趋势衰减', target['execution_reason']) def test_completed_status_for_take_profit(self): self._insert_rec( symbol='DDD/USDT', status='hit_tp1', action_status='止盈1', current_price=111.0, pnl_pct=11.0, max_pnl_pct=12.0, entry_plan_json=json.dumps({ 'entry_price': 100.0, 'entry_action': '可即刻买入', 'risk_reward_ok': True, }, ensure_ascii=False), ) all_rows = altcoin_db.get_all_recommendations(limit=10) target = next(r for r in all_rows if r['symbol'] == 'DDD/USDT') self.assertEqual(target['execution_status'], 'completed') self.assertEqual(target['execution_label'], '✅ 已兑现，仅观察') self.assertIn('止盈', target['execution_reason']) stats = altcoin_db.get_stats() self.assertEqual(stats['active_count'], 0) self.assertIsNone(stats['leaderboard']['top_gainer']) def test_create_recommendation_skips_duplicate_symbol_state_within_window(self): with patch.object(altcoin_db, 'get_meta', return_value={'strategy_version': 'v1.2'}), \ patch.object(altcoin_db, 'datetime') as mock_datetime: mock_datetime.now.return_value = unittest.mock.Mock(isoformat=lambda: '2026-04-30T10:00:00') rec_id_1 = altcoin_db.create_recommendation( symbol='AAA/USDT', rec_state='加速', rec_score=8, entry_price=100.0, sector='AI', signals=['4H 连续4K多头加速'], is_meme=0, direction='多头启动' ) rec_id_2 = altcoin_db.create_recommendation( symbol='AAA/USDT', rec_state='加速', rec_score=9, entry_price=101.0, sector='AI', signals=['4H 连续4K多头加速'], is_meme=0, direction='多头启动' ) self.assertEqual(rec_id_1, rec_id_2) conn = sqlite3.connect(self.db_path) count = conn.execute("SELECT COUNT(*) FROM recommendation WHERE symbol='AAA/USDT'").fetchone()[0] conn.close() self.assertEqual(count, 1) def test_create_recommendation_allows_new_state_transition_within_window(self): with patch.object(altcoin_db, 'get_meta', return_value={'strategy_version': 'v1.2'}), \ patch.object(altcoin_db, 'datetime') as mock_datetime: mock_datetime.now.return_value = unittest.mock.Mock(isoformat=lambda: '2026-04-30T10:00:00') rec_id_1 = altcoin_db.create_recommendation( symbol='AAA/USDT', rec_state='蓄力', rec_score=5, entry_price=100.0, sector='AI', signals=['4H 3静K蓄力'], is_meme=0, direction='多头启动' ) rec_id_2 = altcoin_db.create_recommendation( symbol='AAA/USDT', rec_state='加速', rec_score=9, entry_price=103.0, sector='AI', signals=['4H 连续4K多头加速'], is_meme=0, direction='多头启动' ) self.assertNotEqual(rec_id_1, rec_id_2) conn = sqlite3.connect(self.db_path) count = conn.execute("SELECT COUNT(*) FROM recommendation WHERE symbol='AAA/USDT'").fetchone()[0] conn.close() self.assertEqual(count, 2) def test_risk_reward_false_blocks_buy_now(self): self._insert_rec( symbol='EEE/USDT', action_status='可即刻买入', current_price=0.0758, entry_price=0.0758, signals=json.dumps(['1H 起爆点↑(强度56×)', '⚠️ 等回踩降权(-3分)'], ensure_ascii=False), entry_plan_json=json.dumps({ 'entry_price': 0.072, 'entry_action': '等回踩', 'risk_reward_ok': False, 'rr1': 0.4, 'stop_loss': 0.07, }, ensure_ascii=False), ) rows = altcoin_db.get_active_recommendations_deduped(actionable_only=False) target = next(r for r in rows if r['symbol'] == 'EEE/USDT') self.assertNotEqual(target['execution_status'], 'buy_now') self.assertIn(target['action_status'], ('等回踩', '观察')) self.assertIn('entry_quality_gate', target['entry_plan']) def test_update_action_status_refuses_buy_now_when_rr_bad(self): self._insert_rec( symbol='FFF/USDT', action_status='持有', current_price=0.0758, entry_price=0.0758, entry_plan_json=json.dumps({ 'entry_price': 0.072, 'entry_action': '等回踩', 'risk_reward_ok': False, 'rr1': 0.4, }, ensure_ascii=False), ) conn = sqlite3.connect(self.db_path) rec_id = conn.execute("SELECT id FROM recommendation WHERE symbol='FFF/USDT'").fetchone()[0] conn.close() altcoin_db.update_recommendation_action_status(rec_id, '可即刻买入') conn = sqlite3.connect(self.db_path) row = conn.execute("SELECT action_status, entry_plan_json FROM recommendation WHERE id=?", (rec_id,)).fetchone() conn.close() self.assertNotEqual(row[0], '可即刻买入') self.assertIn(row[0], ('等回踩', '观察')) self.assertIn('entry_quality_gate', json.loads(row[1])) def test_create_recommendation_stores_stable_signal_codes(self): rec_id = altcoin_db.create_recommendation( symbol='SIG/USDT', rec_state='爆发', rec_score=12, entry_price=1.0, signals=['1H 量价齐飞K(量3.7x)', '1H 量价齐飞K(量9.1x)', '15min 回踩确认'], entry_plan={'entry_action': '可即刻买入', 'entry_price': 1.0, 'risk_reward_ok': True, 'rr1': 2.0}, direction='多头启动', ) conn = sqlite3.connect(self.db_path) row = conn.execute("SELECT signal_codes_json, signal_labels_json FROM recommendation WHERE id=?", (rec_id,)).fetchone() conn.close() codes = json.loads(row[0]) labels = json.loads(row[1]) self.assertEqual(codes.count('vp_fly_1h_current'), 1) self.assertIn('pullback_15m_confirm', codes) self.assertEqual(len(labels), 3) def test_closed_paper_trade_puts_symbol_into_cooling_off(self): self._insert_rec( symbol='HYPER/USDT', action_status='等回踩', current_price=0.1257, entry_plan_json=json.dumps({ 'entry_price': 0.109065, 'entry_action': '等回踩', 'stop_loss': 0.118914, 'tp1': 0.135879, 'risk_reward_ok': True, 'rr1': 1.5, }, ensure_ascii=False), ) conn = sqlite3.connect(self.db_path) conn.execute( ''' INSERT INTO paper_trades ( recommendation_id, symbol, side, status, opened_at, closed_at, entry_price, exit_price, qty, notional_usdt, stop_loss, tp1, tp2, max_price, min_price, current_price, pnl_pct, realized_pnl_pct, realized_pnl_usdt, fee_usdt, exit_reason, source_status, source_action, strategy_version, created_at, updated_at ) VALUES (?, ?, 'long', 'closed', ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ''', ( 1, 'HYPER/USDT', '2026-05-18T14:41:09', '2026-05-18T15:11:03', 0.11275635, 0.1141429, 44343.40061557508, 5000.0, 0.10906, 0.118714, 0.12165, 0.1163, 0.1125, 0.1141429, 1.2297, 1.2297, 51.485, 10.0, 'trailing_stop', 'buy_now', '可即刻买入', 'v-test', '2026-05-18T14:41:09', '2026-05-18T15:11:03' ), ) conn.commit() conn.close() rows = altcoin_db.get_active_recommendations_deduped(actionable_only=False) target = next(r for r in rows if r['symbol'] == 'HYPER/USDT') self.assertIn(target['action_status'], ('观察', '等回踩', '止盈1', '止盈2', '止损', '跟踪止盈')) self.assertFalse(target['execution_status'] == 'buy_now' and target['entry_triggered'] == 1) if __name__ == '__main__': unittest.main()