Codex
255 lines
9.7 KiB
Python
255 lines
9.7 KiB
Python
from datetime import date, datetime, timezone
|
|
import os
|
|
import unittest
|
|
|
|
os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://user:pass@localhost:5432/ai_ict_test")
|
|
|
|
from scripts import run_trading_plans
|
|
from src.services.trading_plans import ACTIVE_PLAN_STATUS, DRAFT_PLAN_STATUS, TradingPlanService
|
|
from src.services.trading_plans.trading_plan_service import build_trading_plan_from_inputs
|
|
|
|
|
|
class TradingPlanServiceTests(unittest.TestCase):
|
|
def test_builds_active_plan_when_draw_and_invalidation_exist(self) -> None:
|
|
plan = build_trading_plan_from_inputs(
|
|
instrument_id=1,
|
|
plan_date=date(2026, 5, 6),
|
|
timeframe_set={"bias_tf": "1h", "setup_tf": "15m", "trigger_tf": "5m"},
|
|
range_row=_range_row(),
|
|
liquidity_pools=[
|
|
_liquidity_pool(pool_id=21, side="buy_side"),
|
|
_liquidity_pool(pool_id=22, side="sell_side"),
|
|
],
|
|
mss=_mss_row(direction="bullish", accepted=True),
|
|
fvgs=[_fvg_row()],
|
|
sessions=[_session_row("LONDON"), _session_row("NY_AM"), _session_row("ASIA")],
|
|
created_ts=_created_ts(),
|
|
)
|
|
|
|
self.assertEqual(plan.status, ACTIVE_PLAN_STATUS)
|
|
self.assertEqual(plan.bias, "bullish")
|
|
self.assertEqual(plan.primary_draw["side"], "buy_side")
|
|
self.assertEqual(plan.primary_draw["source"], "liquidity_pool")
|
|
self.assertEqual(plan.primary_draw["pool_id"], 21)
|
|
self.assertEqual(plan.allowed_sessions, ["LONDON", "NY_AM"])
|
|
self.assertTrue(plan.invalidations)
|
|
self.assertEqual(plan.no_trade_conditions, [])
|
|
self.assertTrue(plan.quality["has_draw"])
|
|
self.assertTrue(plan.quality["has_invalidation"])
|
|
self.assertTrue(plan.quality["active_eligible"])
|
|
self.assertEqual(plan.ict_rulebook_mapping["surface"], "trading_plan")
|
|
self.assertEqual(plan.ict_rulebook_mapping["status"], "mapped")
|
|
self.assertIn("draw_on_liquidity", plan.ict_rulebook_mapping["present_concepts"])
|
|
self.assertIn("trading_plan", plan.ict_rulebook_mapping["present_concepts"])
|
|
self.assertIsNotNone(plan.activated_ts)
|
|
|
|
def test_plan_without_draw_cannot_be_active_even_when_invalidation_exists(self) -> None:
|
|
plan = build_trading_plan_from_inputs(
|
|
instrument_id=1,
|
|
plan_date=date(2026, 5, 6),
|
|
timeframe_set={"bias_tf": "1h", "setup_tf": "15m", "trigger_tf": "5m"},
|
|
range_row=None,
|
|
liquidity_pools=[],
|
|
mss=_mss_row(direction="bullish", accepted=True),
|
|
fvgs=[],
|
|
sessions=[],
|
|
created_ts=_created_ts(),
|
|
)
|
|
|
|
self.assertEqual(plan.status, DRAFT_PLAN_STATUS)
|
|
self.assertEqual(plan.primary_draw, {})
|
|
self.assertTrue(plan.invalidations)
|
|
self.assertIn("missing_primary_draw", plan.no_trade_conditions)
|
|
self.assertFalse(plan.quality["active_eligible"])
|
|
self.assertIsNone(plan.activated_ts)
|
|
|
|
def test_plan_without_invalidation_cannot_be_active_even_when_draw_exists(self) -> None:
|
|
plan = build_trading_plan_from_inputs(
|
|
instrument_id=1,
|
|
plan_date=date(2026, 5, 6),
|
|
timeframe_set={"bias_tf": "1h", "setup_tf": "15m", "trigger_tf": "5m"},
|
|
range_row=None,
|
|
liquidity_pools=[_liquidity_pool(pool_id=41, side="buy_side")],
|
|
mss=None,
|
|
fvgs=[],
|
|
sessions=[],
|
|
created_ts=_created_ts(),
|
|
)
|
|
|
|
self.assertEqual(plan.status, DRAFT_PLAN_STATUS)
|
|
self.assertEqual(plan.primary_draw["side"], "buy_side")
|
|
self.assertEqual(plan.invalidations, [])
|
|
self.assertIn("missing_invalidation", plan.no_trade_conditions)
|
|
self.assertIn("missing_directional_bias", plan.no_trade_conditions)
|
|
self.assertFalse(plan.quality["active_eligible"])
|
|
|
|
def test_mss_must_be_accepted_for_active_plan(self) -> None:
|
|
plan = build_trading_plan_from_inputs(
|
|
instrument_id=1,
|
|
plan_date=date(2026, 5, 6),
|
|
timeframe_set={"bias_tf": "1h", "setup_tf": "15m", "trigger_tf": "5m"},
|
|
range_row=_range_row(),
|
|
liquidity_pools=[_liquidity_pool(pool_id=51, side="sell_side")],
|
|
mss=_mss_row(direction="bearish", accepted=False),
|
|
fvgs=[_fvg_row(direction="bearish")],
|
|
sessions=[],
|
|
created_ts=_created_ts(),
|
|
)
|
|
|
|
self.assertEqual(plan.status, DRAFT_PLAN_STATUS)
|
|
self.assertEqual(plan.bias, "bearish")
|
|
self.assertIn("mss_not_accepted", plan.no_trade_conditions)
|
|
|
|
def test_default_btc_eth_symbols_can_generate_structured_daily_plans(self) -> None:
|
|
repository = _TradingPlanStubRepository()
|
|
service = TradingPlanService(repository=repository)
|
|
plans = []
|
|
|
|
for symbol in run_trading_plans.build_trading_plan_symbols_from_env({}):
|
|
instrument = repository.fetch_instrument_by_symbol(symbol)
|
|
self.assertIsNotNone(instrument)
|
|
assert instrument is not None
|
|
plans.append(
|
|
service.build_plan(
|
|
instrument_id=instrument["id"],
|
|
bias_tf="1h",
|
|
setup_tf="15m",
|
|
trigger_tf="5m",
|
|
plan_date=date(2026, 5, 6),
|
|
)
|
|
)
|
|
|
|
self.assertEqual([plan.instrument_id for plan in plans], [1, 2])
|
|
self.assertEqual([plan.status for plan in plans], [ACTIVE_PLAN_STATUS, ACTIVE_PLAN_STATUS])
|
|
for plan in plans:
|
|
with self.subTest(instrument_id=plan.instrument_id):
|
|
self.assertTrue(plan.primary_draw)
|
|
self.assertTrue(plan.invalidations)
|
|
self.assertIn(plan.bias, {"bullish", "bearish"})
|
|
|
|
|
|
class TradingPlanScriptConfigTests(unittest.TestCase):
|
|
def test_symbol_env_defaults_to_btc_eth(self) -> None:
|
|
self.assertEqual(run_trading_plans.build_trading_plan_symbols_from_env({}), ("BTC-USDT", "ETH-USDT"))
|
|
|
|
def test_symbol_env_accepts_comma_separated_override(self) -> None:
|
|
self.assertEqual(
|
|
run_trading_plans.build_trading_plan_symbols_from_env({"TRADING_PLAN_SYMBOLS": " BTC-USDT , ETH-USDT "}),
|
|
("BTC-USDT", "ETH-USDT"),
|
|
)
|
|
|
|
def test_timeframe_env_reads_overrides(self) -> None:
|
|
self.assertEqual(
|
|
run_trading_plans.build_trading_plan_timeframe_set_from_env(
|
|
{
|
|
"AI_ICT_BIAS_TIMEFRAME": "4h",
|
|
"AI_ICT_SETUP_TIMEFRAME": "1h",
|
|
"AI_ICT_TRIGGER_TIMEFRAME": "15m",
|
|
}
|
|
),
|
|
{"bias_tf": "4h", "setup_tf": "1h", "trigger_tf": "15m"},
|
|
)
|
|
|
|
|
|
class _TradingPlanStubRepository:
|
|
def __init__(self) -> None:
|
|
self.instruments = {
|
|
"BTC-USDT": {"id": 1, "symbol": "BTC-USDT", "active": True},
|
|
"ETH-USDT": {"id": 2, "symbol": "ETH-USDT", "active": True},
|
|
}
|
|
|
|
def fetch_instrument_by_symbol(self, symbol: str):
|
|
return self.instruments.get(symbol)
|
|
|
|
def fetch_latest_range(self, instrument_id: int, timeframe: str):
|
|
if instrument_id == 1:
|
|
return _range_row(range_id=101, high=72000.0, low=70000.0)
|
|
return _range_row(range_id=201, high=3800.0, low=3600.0)
|
|
|
|
def fetch_recent_liquidity_pools(self, instrument_id: int, timeframe: str, limit: int = 20):
|
|
if instrument_id == 1:
|
|
return [_liquidity_pool(pool_id=111, side="buy_side", price_low=71980.0, price_high=72020.0)]
|
|
return [_liquidity_pool(pool_id=211, side="sell_side", price_low=3590.0, price_high=3610.0)]
|
|
|
|
def fetch_latest_mss(self, instrument_id: int, timeframe: str):
|
|
if instrument_id == 1:
|
|
return _mss_row(mss_id=121, direction="bullish", accepted=True, broken_level=71000.0)
|
|
return _mss_row(mss_id=221, direction="bearish", accepted=True, broken_level=3700.0)
|
|
|
|
def fetch_recent_fvgs(self, instrument_id: int, timeframe: str, limit: int = 20):
|
|
if instrument_id == 1:
|
|
return [_fvg_row(fvg_id=131, direction="bullish")]
|
|
return [_fvg_row(fvg_id=231, direction="bearish")]
|
|
|
|
def fetch_sessions_for_date(self, instrument_id: int, plan_date):
|
|
return [_session_row("LONDON"), _session_row("NY_AM")]
|
|
|
|
def upsert_trading_plan(self, **kwargs):
|
|
return int(kwargs["instrument_id"])
|
|
|
|
|
|
def _created_ts() -> datetime:
|
|
return datetime(2026, 5, 6, 0, 0, tzinfo=timezone.utc)
|
|
|
|
|
|
def _range_row(range_id: int = 10, high: float = 101.0, low: float = 99.0) -> dict:
|
|
return {"id": range_id, "high": high, "low": low, "equilibrium": (high + low) / 2, "state": "valid"}
|
|
|
|
|
|
def _liquidity_pool(
|
|
*,
|
|
pool_id: int,
|
|
side: str,
|
|
price_low: float = 100.8,
|
|
price_high: float = 101.2,
|
|
status: str = "untouched",
|
|
) -> dict:
|
|
return {
|
|
"id": pool_id,
|
|
"side": side,
|
|
"pool_type": "equal_highs" if side == "buy_side" else "equal_lows",
|
|
"price_low": price_low,
|
|
"price_high": price_high,
|
|
"status": status,
|
|
}
|
|
|
|
|
|
def _mss_row(
|
|
*,
|
|
mss_id: int = 30,
|
|
direction: str = "bullish",
|
|
accepted: bool = True,
|
|
broken_level: float = 100.0,
|
|
) -> dict:
|
|
return {
|
|
"id": mss_id,
|
|
"direction": direction,
|
|
"accepted": accepted,
|
|
"broken_level": broken_level,
|
|
"displacement_event_id": 7,
|
|
"meta": {"state": "accepted" if accepted else "rejected"},
|
|
}
|
|
|
|
|
|
def _fvg_row(fvg_id: int = 40, direction: str = "bullish") -> dict:
|
|
return {
|
|
"id": fvg_id,
|
|
"direction": direction,
|
|
"lower": 100.0,
|
|
"upper": 100.2,
|
|
"status": "open",
|
|
"fill_pct": 0.0,
|
|
}
|
|
|
|
|
|
def _session_row(code: str) -> dict:
|
|
return {
|
|
"id": len(code),
|
|
"session_code": code,
|
|
"status": "planned",
|
|
}
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|