import json import os import sqlite3 import tempfile import unittest from unittest.mock import patch 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='2026-04-29T10:00:00', rec_state='加速', rec_score=8, entry_price=100.0, stop_loss=95.0, tp1=110.0, tp2=118.0, sector='AI', signals='[]', 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='2026-04-29T10:05:00', 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, }, 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_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_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])) if __name__ == '__main__': unittest.main()