40 lines
1.9 KiB
Python
40 lines
1.9 KiB
Python
import json
|
|
|
|
from app.db.recommendation_commands import downgrade_active_entries_for_market_risk
|
|
|
|
|
|
def test_downgrade_active_entries_for_market_risk_updates_executable_recs(pg_conn):
|
|
pg_conn.execute(
|
|
"""
|
|
INSERT INTO recommendation (
|
|
symbol, rec_time, rec_state, rec_score, entry_price, status,
|
|
action_status, execution_status, display_bucket, lifecycle_state,
|
|
entry_plan_json
|
|
)
|
|
VALUES
|
|
('BUY/USDT', '2026-05-26T10:00:00', '爆发', 30, 1.0, 'active',
|
|
'可即刻买入', 'buy_now', 'realtime', 'buyable', '{"entry_action":"可即刻买入"}'),
|
|
('WAIT/USDT', '2026-05-26T10:01:00', '爆发', 20, 1.0, 'active',
|
|
'等回踩', 'wait_pullback', 'watch_pool', 'waiting_entry', '{"entry_action":"等回踩"}'),
|
|
('OBS/USDT', '2026-05-26T10:02:00', '观察', 10, 1.0, 'active',
|
|
'观察', 'observe', 'watch_pool', 'watching', '{"entry_action":"观察"}')
|
|
"""
|
|
)
|
|
pg_conn.commit()
|
|
|
|
result = downgrade_active_entries_for_market_risk("critical test", event_time="2026-05-26T11:00:00")
|
|
|
|
assert result["updated_count"] == 2
|
|
rows = pg_conn.execute(
|
|
"SELECT symbol, action_status, execution_status, display_bucket, entry_plan_json FROM recommendation ORDER BY symbol"
|
|
).fetchall()
|
|
by_symbol = {row["symbol"]: dict(row) for row in rows}
|
|
assert by_symbol["BUY/USDT"]["action_status"] == "观察"
|
|
assert by_symbol["BUY/USDT"]["execution_status"] == "observe"
|
|
assert by_symbol["WAIT/USDT"]["action_status"] == "观察"
|
|
assert by_symbol["WAIT/USDT"]["execution_status"] == "observe"
|
|
assert by_symbol["OBS/USDT"]["action_status"] == "观察"
|
|
plan = json.loads(by_symbol["BUY/USDT"]["entry_plan_json"])
|
|
assert plan["market_risk_gate"]["blocked_action"] == "可即刻买入"
|
|
assert plan["market_risk_gate"]["final_action"] == "观察"
|