187 lines
6.6 KiB
Python
187 lines
6.6 KiB
Python
import json
|
|
import os
|
|
import sqlite3
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
|
|
from app.db import altcoin_db
|
|
|
|
|
|
class RecommendationStateMainlineTests(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='CHIP/USDT',
|
|
rec_time='2026-05-09T20:10:21',
|
|
rec_state='爆发',
|
|
rec_score=27,
|
|
entry_price=0.06557,
|
|
stop_loss=0.061846,
|
|
tp1=0.072000,
|
|
tp2=0.074881,
|
|
sector='',
|
|
signals=json.dumps(['15min 回踩确认'], ensure_ascii=False),
|
|
is_meme=0,
|
|
status='active',
|
|
current_price=0.06557,
|
|
max_price=0.06557,
|
|
min_price=0.06557,
|
|
pnl_pct=0.0,
|
|
max_pnl_pct=0.0,
|
|
max_drawdown_pct=0.0,
|
|
hit_tp1_time='',
|
|
hit_tp2_time='',
|
|
stopped_out_time='',
|
|
expired_time='',
|
|
last_track_time='2026-05-09T20:11:00',
|
|
entry_plan_json=json.dumps({
|
|
'entry_price': 0.06557,
|
|
'entry_action': '等回踩',
|
|
'risk_reward_ok': True,
|
|
'rr1': 1.5,
|
|
'stop_loss': 0.061846,
|
|
'tp1': 0.072000,
|
|
'entry_trigger_confirmed': True,
|
|
}, ensure_ascii=False),
|
|
action_status='等回踩',
|
|
direction='多头启动',
|
|
strategy_version='v-test',
|
|
)
|
|
defaults.update(kwargs)
|
|
conn = sqlite3.connect(self.db_path)
|
|
cols = ','.join(defaults.keys())
|
|
placeholders = ','.join(['?'] * len(defaults))
|
|
cur = conn.execute(
|
|
f"INSERT INTO recommendation ({cols}) VALUES ({placeholders})",
|
|
tuple(defaults.values()),
|
|
)
|
|
conn.commit()
|
|
conn.close()
|
|
return cur.lastrowid
|
|
|
|
def _row(self, rec_id):
|
|
conn = sqlite3.connect(self.db_path)
|
|
conn.row_factory = sqlite3.Row
|
|
row = dict(conn.execute('SELECT * FROM recommendation WHERE id=?', (rec_id,)).fetchone())
|
|
conn.close()
|
|
return row
|
|
|
|
def test_state_transition_updates_db_before_push_payload(self):
|
|
rec_id = self._insert_rec()
|
|
|
|
decision = altcoin_db.apply_recommendation_state_transition(
|
|
rec_id,
|
|
requested_action='可即刻买入',
|
|
current_price=0.06580,
|
|
event_time='2026-05-09T22:21:12',
|
|
)
|
|
|
|
self.assertEqual(decision['action_status'], '可即刻买入')
|
|
self.assertEqual(decision['execution_status'], 'buy_now')
|
|
self.assertTrue(decision['push_required'])
|
|
self.assertEqual(decision['push_symbol'], 'CHIP/USDT')
|
|
self.assertEqual(decision['push_entry_price'], 0.06580)
|
|
self.assertEqual(decision['push_current_price'], 0.06580)
|
|
self.assertEqual(decision['push_pnl_pct'], 0.0)
|
|
|
|
row = self._row(rec_id)
|
|
self.assertEqual(row['action_status'], '可即刻买入')
|
|
self.assertEqual(row['entry_price'], 0.06580)
|
|
self.assertEqual(row['current_price'], 0.06580)
|
|
self.assertEqual(row['pnl_pct'], 0.0)
|
|
self.assertEqual(row['rec_time'], '2026-05-09T22:21:12')
|
|
|
|
def test_state_transition_blocks_push_when_gate_downgrades_action(self):
|
|
rec_id = self._insert_rec(
|
|
entry_plan_json=json.dumps({
|
|
'entry_price': 0.06557,
|
|
'entry_action': '等回踩',
|
|
'risk_reward_ok': False,
|
|
'rr1': 0.4,
|
|
'stop_loss': 0.061846,
|
|
'tp1': 0.066,
|
|
}, ensure_ascii=False),
|
|
current_price=0.06650,
|
|
)
|
|
|
|
decision = altcoin_db.apply_recommendation_state_transition(
|
|
rec_id,
|
|
requested_action='可即刻买入',
|
|
current_price=0.06650,
|
|
event_time='2026-05-09T22:21:12',
|
|
)
|
|
|
|
self.assertNotEqual(decision['action_status'], '可即刻买入')
|
|
self.assertFalse(decision['push_required'])
|
|
row = self._row(rec_id)
|
|
self.assertNotEqual(row['action_status'], '可即刻买入')
|
|
self.assertNotEqual(row['rec_time'], '2026-05-09T22:21:12')
|
|
|
|
def test_state_transition_merges_tracker_signal_with_persisted_signals(self):
|
|
rec_id = self._insert_rec(
|
|
action_status='等回踩',
|
|
entry_price=0.114,
|
|
current_price=0.114,
|
|
entry_plan_json=json.dumps({
|
|
'entry_price': 0.114,
|
|
'entry_action': '等回踩',
|
|
'risk_reward_ok': True,
|
|
'rr1': 2.35,
|
|
'stop_loss': 0.1026,
|
|
'tp1': 0.140743,
|
|
'opportunity_level': 'structure_watch',
|
|
'opportunity_level_label': '结构观察',
|
|
'max_action': 'wait_pullback',
|
|
}, ensure_ascii=False),
|
|
signals=json.dumps(['4H需求区反弹'], ensure_ascii=False),
|
|
)
|
|
|
|
decision = altcoin_db.apply_recommendation_state_transition(
|
|
rec_id,
|
|
requested_action='可即刻买入',
|
|
current_price=0.114,
|
|
event_time='2026-05-09T22:21:12',
|
|
signals=['🟢 回踩确认完毕！可即刻入场(15min动K确认)'],
|
|
)
|
|
|
|
self.assertEqual(decision['action_status'], '可即刻买入')
|
|
self.assertEqual(decision['execution_status'], 'buy_now')
|
|
self.assertTrue(decision['push_required'])
|
|
|
|
def test_api_derivation_consumes_persisted_state_without_promoting_initial_action(self):
|
|
self._insert_rec(
|
|
symbol='AAA/USDT',
|
|
action_status='等回踩',
|
|
current_price=104.0,
|
|
entry_price=100.0,
|
|
entry_plan_json=json.dumps({
|
|
'entry_price': 100.0,
|
|
'entry_action': '可即刻买入',
|
|
'risk_reward_ok': True,
|
|
'rr1': 2.0,
|
|
}, ensure_ascii=False),
|
|
)
|
|
|
|
rows = altcoin_db.get_active_recommendations_deduped(actionable_only=False, version='v-test')
|
|
target = next(r for r in rows if r['symbol'] == 'AAA/USDT')
|
|
self.assertEqual(target['action_status'], '等回踩')
|
|
self.assertEqual(target['execution_status'], 'wait_pullback')
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|