yuqei/ai-exchange
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Phase 4 target selection and exits

Browse Source
This commit is contained in:
Codex
2026-05-06 13:41:48 +08:00
parent 7a569daecd
commit 3b2d1636e2
16 changed files with 990 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files
README.md
+3 -3
View File
@@ -219,10 +219,10 @@ Record the current commit as the last-known-good SHA before deploy. If any gate
- `python3 scripts/run_fvg.py` - rebuild FVG zones
- `python3 scripts/run_trading_plans.py` - generate today's structured BTC/ETH trading plans before signal generation; active plans require a Draw and at least one Invalidation
- `python3 scripts/run_trade_plan_candidates.py` - watch the active trading plan for the first executable candidate model: OSOK/Daily Range context, Sweep, MSS, and FVG retrace; missing confirmations are reported instead of promoted to execution
- `python3 scripts/run_signals.py` - generate bias and signals
- `python3 scripts/run_execution.py` - run paper execution gating, create an execution record, then auto-settle the latest execution for that signal
- `python3 scripts/run_signals.py` - generate bias and signals with structural `target_plan` selection; TP1 prefers IRL and TP2 prefers ERL, and signals are skipped when traceable structural targets are missing
- `python3 scripts/run_execution.py` - run paper execution gating, require structural targets plus management rules for modeled signals, create an execution record, then auto-settle the latest execution for that signal
- `python3 scripts/run_risk_governor.py` - run the independent market-state and risk governor pre-execution gate
- `python3 scripts/run_backtest.py` - run backtest and write results
- `python3 scripts/run_backtest.py` - run backtest and write results, including target-source traceability and management rules in result meta
- `python3 scripts/run_autonomous_cycle.py` - run the unattended cycle: offline checks, database probe, readiness checks, then guarded pipeline when the database is available; prints `Next actions` for the next unattended step
- `python3 scripts/run_database_readiness.py` - run read-only database readiness checks for connection, schema, seed data, and candle availability; prints `next_actions` and `action_plan` in JSON mode
- `.venv/bin/python scripts/run_trading_delivery_rehearsal.py` - run the P0 trading delivery rehearsal as one scripted flow: checklist drift check, historical repair decision, readiness/auto-backfill, guarded pipeline, deploy/health-check, and post-deploy observe
package.json
+1 -1
View File
@@ -3,7 +3,7 @@
"version": "0.1.0",
"private": true,
"scripts": {
"lint": ".\\.venv\\Scripts\\python.exe scripts/run_script_static_checks.py && .\\.venv\\Scripts\\python.exe scripts/run_migration_static_checks.py && .\\.venv\\Scripts\\python.exe -m unittest tests.unit.test_trading_plan_service tests.unit.test_market_state_and_risk_governor tests.unit.test_trade_plan_candidate_service tests.unit.test_execution_service tests.unit.test_api_server",
"lint": ".\\.venv\\Scripts\\python.exe scripts/run_script_static_checks.py && .\\.venv\\Scripts\\python.exe scripts/run_migration_static_checks.py && .\\.venv\\Scripts\\python.exe -m unittest tests.unit.test_trading_plan_service tests.unit.test_market_state_and_risk_governor tests.unit.test_target_selection_service tests.unit.test_trade_plan_candidate_service tests.unit.test_execution_service tests.unit.test_api_server",
"build": ".\\.venv\\Scripts\\python.exe -m compileall -q src scripts tests",
"visual:snapshots": "node --check src/web/app.js"
}
src/repositories/backtest_repository.py
+1
View File
@@ -31,6 +31,7 @@ class BacktestRepository:
ts.rr_tp2,
ts.created_ts,
ts.expires_ts,
ts.meta as signal_meta,
i.symbol,
bs.bias_tf,
bs.setup_tf,
src/repositories/signal_repository.py
+69
View File
@@ -136,6 +136,75 @@ class SignalRepository:
).mappings().first()
return dict(row) if row else None
def fetch_latest_range(self, instrument_id: int, timeframe: str) -> Optional[dict]:
sql = text(
"""
select id, high, low, equilibrium, state, ts_start, ts_end, meta
from dealing_ranges
where instrument_id = :instrument_id
and timeframe = :timeframe
and state = 'valid'
order by ts_start desc, id desc
limit 1
"""
)
with SessionLocal() as session:
row = session.execute(sql, {"instrument_id": instrument_id, "timeframe": timeframe}).mappings().first()
return dict(row) if row else None
def fetch_recent_liquidity_pools(self, instrument_id: int, timeframe: str, limit: int = 50) -> list[dict]:
sql = text(
"""
select id, side, pool_type, price_low, price_high, status, created_ts, updated_ts, meta
from liquidity_pools
where instrument_id = :instrument_id
and timeframe = :timeframe
order by
case when status in ('untouched', 'active', 'open') then 0 else 1 end,
created_ts desc,
id desc
limit :limit
"""
)
with SessionLocal() as session:
rows = session.execute(
sql,
{"instrument_id": instrument_id, "timeframe": timeframe, "limit": limit},
).mappings().all()
return [dict(row) for row in rows]
def fetch_recent_sessions_for_targets(self, instrument_id: int, limit: int = 10) -> list[dict]:
sql = text(
"""
select id, session_code, status, start_ts, end_ts, high, low, midpoint, meta
from market_sessions
where instrument_id = :instrument_id
order by start_ts desc, id desc
limit :limit
"""
)
with SessionLocal() as session:
rows = session.execute(sql, {"instrument_id": instrument_id, "limit": limit}).mappings().all()
return [dict(row) for row in rows]
def fetch_recent_candles_for_targets(self, instrument_id: int, timeframe: str, limit: int = 100) -> list[dict]:
sql = text(
"""
select ts_open, ts_close, open, high, low, close
from candles
where instrument_id = :instrument_id
and timeframe = :timeframe
order by ts_open desc
limit :limit
"""
)
with SessionLocal() as session:
rows = session.execute(
sql,
{"instrument_id": instrument_id, "timeframe": timeframe, "limit": limit},
).mappings().all()
return [dict(row) for row in rows]
def insert_trade_signal(
self,
instrument_id: int,
src/repositories/trade_plan_candidate_repository.py
+69
View File
@@ -110,6 +110,75 @@ class TradePlanCandidateRepository:
).mappings().all()
return [dict(row) for row in rows]
def fetch_latest_range(self, instrument_id: int, timeframe: str) -> Optional[dict]:
sql = text(
"""
select id, high, low, equilibrium, state, ts_start, ts_end, meta
from dealing_ranges
where instrument_id = :instrument_id
and timeframe = :timeframe
and state = 'valid'
order by ts_start desc, id desc
limit 1
"""
)
with SessionLocal() as session:
row = session.execute(sql, {"instrument_id": instrument_id, "timeframe": timeframe}).mappings().first()
return dict(row) if row else None
def fetch_recent_liquidity_pools(self, instrument_id: int, timeframe: str, limit: int = 50) -> list[dict]:
sql = text(
"""
select id, side, pool_type, price_low, price_high, status, created_ts, updated_ts, meta
from liquidity_pools
where instrument_id = :instrument_id
and timeframe = :timeframe
order by
case when status in ('untouched', 'active', 'open') then 0 else 1 end,
created_ts desc,
id desc
limit :limit
"""
)
with SessionLocal() as session:
rows = session.execute(
sql,
{"instrument_id": instrument_id, "timeframe": timeframe, "limit": limit},
).mappings().all()
return [dict(row) for row in rows]
def fetch_recent_sessions_for_targets(self, instrument_id: int, limit: int = 10) -> list[dict]:
sql = text(
"""
select id, session_code, status, start_ts, end_ts, high, low, midpoint, meta
from market_sessions
where instrument_id = :instrument_id
order by start_ts desc, id desc
limit :limit
"""
)
with SessionLocal() as session:
rows = session.execute(sql, {"instrument_id": instrument_id, "limit": limit}).mappings().all()
return [dict(row) for row in rows]
def fetch_recent_candles_for_targets(self, instrument_id: int, timeframe: str, limit: int = 100) -> list[dict]:
sql = text(
"""
select ts_open, ts_close, open, high, low, close
from candles
where instrument_id = :instrument_id
and timeframe = :timeframe
order by ts_open desc
limit :limit
"""
)
with SessionLocal() as session:
rows = session.execute(
sql,
{"instrument_id": instrument_id, "timeframe": timeframe, "limit": limit},
).mappings().all()
return [dict(row) for row in rows]
def fetch_latest_candidate(self, trading_plan_id: int, model_code: str) -> Optional[dict]:
sql = text(
"""
src/services/backtest/backtest_service.py
+23
View File
@@ -48,6 +48,8 @@ class BacktestSignal:
tp2: float
rr_tp1: float = 0.0
rr_tp2: float = 0.0
target_plan: Optional[dict] = None
management_rules: Optional[list[dict]] = None
@dataclass(frozen=True)
@@ -127,6 +129,8 @@ class BacktestService:
limit=500,
)
if not future:
target_plan = extract_target_plan_from_signal_row(signal)
management_rules = extract_management_rules_from_signal_row(signal)
record = BacktestOutcome(
signal_id=signal['id'],
outcome='missed',
@@ -157,6 +161,8 @@ class BacktestService:
'signal_created_ts': signal['created_ts'],
'rr_tp1': float(signal.get('rr_tp1') or 0.0),
'rr_tp2': float(signal.get('rr_tp2') or 0.0),
'target_plan': target_plan,
'management_rules': management_rules,
'cost_r': 0.0,
'cost_model': cost_config.__dict__,
'backtest_mode': mode,
@@ -193,6 +199,8 @@ class BacktestService:
tp2=float(signal['tp2'] or 0),
rr_tp1=float(signal.get('rr_tp1') or 0),
rr_tp2=float(signal.get('rr_tp2') or 0),
target_plan=extract_target_plan_from_signal_row(signal),
management_rules=extract_management_rules_from_signal_row(signal),
),
future,
cost_config=cost_config,
@@ -221,6 +229,8 @@ class BacktestService:
'signal_created_ts': signal['created_ts'],
'rr_tp1': float(signal.get('rr_tp1') or 0.0),
'rr_tp2': float(signal.get('rr_tp2') or 0.0),
'target_plan': extract_target_plan_from_signal_row(signal),
'management_rules': extract_management_rules_from_signal_row(signal),
'weekday': _extract_weekday(signal['created_ts']),
}
)
@@ -470,6 +480,8 @@ def _evaluate_signal_outcome_steps(
'signal_side': signal.side,
'rr_tp1': signal.rr_tp1,
'rr_tp2': signal.rr_tp2,
'target_plan': dict(signal.target_plan or {}),
'management_rules': list(signal.management_rules or []),
'mfe_r': mfe_r,
'mae_r': mae_r,
'cost_r': cost_r,
@@ -694,6 +706,17 @@ def normalize_timeframe_set(timeframe_set: Optional[dict]) -> dict:
}
def extract_target_plan_from_signal_row(signal: dict) -> dict:
meta = dict(signal.get('signal_meta') or signal.get('meta') or {})
return dict(meta.get('target_plan') or {})
def extract_management_rules_from_signal_row(signal: dict) -> list[dict]:
meta = dict(signal.get('signal_meta') or signal.get('meta') or {})
target_plan = dict(meta.get('target_plan') or {})
return list(meta.get('management_rules') or target_plan.get('management_rules') or [])
def build_timeframe_bucket(*, bias_tf: str, setup_tf: str, trigger_tf: str) -> str:
return f"{bias_tf}/{setup_tf}/{trigger_tf}"
src/services/execution/execution_service.py
+15
View File
@@ -13,6 +13,7 @@ from src.services.backtest import (
classify_backtest_lifecycle_state_from_result,
)
from src.services.risk import AccountRiskState, RiskContext, RiskGovernorService
from src.services.targets import has_complete_structural_target_plan
ALLOWED_EXECUTION_SIGNAL_STATUSES = {"awaiting_confirmation"}
@@ -371,6 +372,9 @@ def evaluate_execution_decision(
risk_context: Optional[RiskContext] = None,
) -> ExecutionDecision:
reasons: list[str] = []
signal_meta = dict(signal.get("meta") or {})
target_plan = dict(signal_meta.get("target_plan") or {})
management_rules = list(signal_meta.get("management_rules") or target_plan.get("management_rules") or [])
meta = {
"signal_side": signal["side"],
"session_code": signal.get("session_code"),
@@ -380,6 +384,8 @@ def evaluate_execution_decision(
"stop_loss": float(signal.get("stop_loss") or 0.0),
"tp1": float(signal.get("tp1") or 0.0),
"tp2": float(signal.get("tp2") or 0.0),
"target_plan": target_plan,
"management_rules": management_rules,
"signal_status": signal.get("status"),
"risk_checks": [],
}
@@ -390,6 +396,10 @@ def evaluate_execution_decision(
reasons.append("signal_not_executable")
if not signal.get("stop_loss"):
reasons.append("missing_stop_loss")
if signal_requires_structural_targets(signal) and not has_complete_structural_target_plan(meta["target_plan"]):
reasons.append("missing_structural_targets")
if signal_requires_structural_targets(signal) and not meta["management_rules"]:
reasons.append("missing_management_rules")
if signal.get("session_code") not in {"LONDON", "NY_AM", "NY_PM"}:
reasons.append("session_not_allowed")
if not signal.get("meta", {}).get("execution_permission", False):
@@ -455,6 +465,11 @@ def evaluate_execution_decision(
)
def signal_requires_structural_targets(signal: dict) -> bool:
meta = dict(signal.get("meta") or {})
return bool(signal.get("model_code")) or bool(meta.get("target_plan_required"))
def evaluate_account_risk_limits(account_state: Optional[AccountRiskState]) -> list[str]:
if account_state is None:
return []
src/services/signals/signal_service.py
+59 -4
View File
@@ -5,6 +5,7 @@ from typing import Optional
from src.config.constants import MIN_TP1_RR, MIN_TP2_RR
from src.repositories.signal_repository import SignalRepository
from src.services.targets import TargetSelectionService, has_complete_structural_target_plan
MINIMAL_SIGNAL_EVIDENCE_KEYS = ("bias_snapshot_id", "fvg_id", "mss_id")
@@ -48,9 +49,11 @@ class SignalService:
self,
repository: Optional[SignalRepository] = None,
required_evidence_keys: tuple[str, ...] = MINIMAL_SIGNAL_EVIDENCE_KEYS,
target_selection_service: Optional[TargetSelectionService] = None,
) -> None:
self.repository = repository or SignalRepository()
self.required_evidence_keys = required_evidence_keys
self.target_selection_service = target_selection_service or TargetSelectionService()
def build_signal(self, instrument_id: int, setup_tf: str, trigger_tf: str) -> Optional[SignalCandidate]:
bias = self.repository.fetch_latest_bias_snapshot(instrument_id=instrument_id, setup_tf=setup_tf, trigger_tf=trigger_tf)
@@ -69,12 +72,21 @@ class SignalService:
entry_high = float(fvg['upper'])
if side == 'buy':
stop_loss = min(float(mss['broken_level']), entry_low) * 0.999
tp1 = entry_high * 1.001
tp2 = entry_high * 1.003
else:
stop_loss = max(float(mss['broken_level']), entry_high) * 1.001
tp1 = entry_low * 0.999
tp2 = entry_low * 0.997
target_plan = self._build_target_plan(
instrument_id=instrument_id,
side=side,
entry_low=entry_low,
entry_high=entry_high,
setup_tf=setup_tf,
trigger_tf=trigger_tf,
)
target_plan_payload = target_plan.to_dict()
if not has_complete_structural_target_plan(target_plan_payload):
return None
tp1 = float(target_plan.tp1.price)
tp2 = float(target_plan.tp2.price)
risk = abs(((entry_low + entry_high) / 2) - stop_loss)
rr_tp1 = abs(tp1 - ((entry_low + entry_high) / 2)) / risk if risk else 0.0
rr_tp2 = abs(tp2 - ((entry_low + entry_high) / 2)) / risk if risk else 0.0
@@ -145,6 +157,9 @@ class SignalService:
'signal_sweep_id': evidence.get('sweep_id'),
'liquidity_target_id': evidence.get('liquidity_target_id'),
'rr_filter': rr_decision,
'target_plan': target_plan_payload,
'management_rules': target_plan_payload['management_rules'],
'target_selection_status': target_plan_payload['status'],
},
)
@@ -197,6 +212,46 @@ class SignalService:
before_ts=before_ts,
)
def _build_target_plan(
self,
*,
instrument_id: int,
side: str,
entry_low: float,
entry_high: float,
setup_tf: str,
trigger_tf: str,
):
range_row = (
self.repository.fetch_latest_range(instrument_id=instrument_id, timeframe=setup_tf)
if hasattr(self.repository, "fetch_latest_range")
else None
)
liquidity_pools = (
self.repository.fetch_recent_liquidity_pools(instrument_id=instrument_id, timeframe=setup_tf, limit=50)
if hasattr(self.repository, "fetch_recent_liquidity_pools")
else []
)
sessions = (
self.repository.fetch_recent_sessions_for_targets(instrument_id=instrument_id, limit=10)
if hasattr(self.repository, "fetch_recent_sessions_for_targets")
else []
)
candles = (
self.repository.fetch_recent_candles_for_targets(instrument_id=instrument_id, timeframe=trigger_tf, limit=100)
if hasattr(self.repository, "fetch_recent_candles_for_targets")
else []
)
return self.target_selection_service.select_targets(
side=side,
entry_low=entry_low,
entry_high=entry_high,
range_row=range_row,
liquidity_pools=liquidity_pools,
sessions=sessions,
candles=candles,
)
def trade_signal_matches_candidate(signal: dict, candidate: SignalCandidate, bias_snapshot_id: int) -> bool:
return (
src/services/targets/__init__.py
+27
View File
@@ -0,0 +1,27 @@
"""Target selection service package."""
from src.services.targets.target_selection_service import (
OBJECTIVE_LAYER_ERL,
OBJECTIVE_LAYER_IRL,
TARGET_PLAN_STATUS_INCOMPLETE,
TARGET_PLAN_STATUS_READY,
TargetPlan,
TargetSelection,
TargetSelectionService,
build_management_rules,
has_complete_structural_target_plan,
target_has_traceable_source,
)
__all__ = [
"OBJECTIVE_LAYER_ERL",
"OBJECTIVE_LAYER_IRL",
"TARGET_PLAN_STATUS_INCOMPLETE",
"TARGET_PLAN_STATUS_READY",
"TargetPlan",
"TargetSelection",
"TargetSelectionService",
"build_management_rules",
"has_complete_structural_target_plan",
"target_has_traceable_source",
]
src/services/targets/target_selection_service.py
+502
View File
@@ -0,0 +1,502 @@
from dataclasses import dataclass
from statistics import mean, pstdev
from typing import Optional
TARGET_PLAN_STATUS_READY = "ready"
TARGET_PLAN_STATUS_INCOMPLETE = "incomplete"
OBJECTIVE_LAYER_IRL = "IRL"
OBJECTIVE_LAYER_ERL = "ERL"
@dataclass(frozen=True)
class TargetSelection:
label: str
objective_layer: str
price: float
source: dict
selection_reason: str
def to_dict(self) -> dict:
return {
"label": self.label,
"objective_layer": self.objective_layer,
"price": self.price,
"source": self.source,
"selection_reason": self.selection_reason,
}
@dataclass(frozen=True)
class TargetPlan:
status: str
side: str
tp1: Optional[TargetSelection]
tp2: Optional[TargetSelection]
management_rules: list[dict]
missing_targets: list[str]
candidates: dict
meta: dict
def to_dict(self) -> dict:
return {
"status": self.status,
"side": self.side,
"tp1": self.tp1.to_dict() if self.tp1 else None,
"tp2": self.tp2.to_dict() if self.tp2 else None,
"management_rules": self.management_rules,
"missing_targets": self.missing_targets,
"candidates": self.candidates,
"meta": self.meta,
}
class TargetSelectionService:
def select_targets(
self,
*,
side: str,
entry_low: float,
entry_high: float,
trading_plan: Optional[dict] = None,
range_row: Optional[dict] = None,
liquidity_pools: Optional[list[dict]] = None,
sessions: Optional[list[dict]] = None,
candles: Optional[list[dict]] = None,
) -> TargetPlan:
entry_price = compute_entry_price(entry_low=entry_low, entry_high=entry_high)
liquidity_pools = list(liquidity_pools or [])
sessions = list(sessions or [])
candles = list(candles or [])
internal_candidates = build_internal_target_candidates(
side=side,
entry_price=entry_price,
range_row=range_row,
liquidity_pools=liquidity_pools,
sessions=sessions,
candles=candles,
)
external_candidates = build_external_target_candidates(
side=side,
entry_price=entry_price,
range_row=range_row,
liquidity_pools=liquidity_pools,
sessions=sessions,
candles=candles,
)
tp1 = internal_candidates[0] if internal_candidates else None
tp2 = external_candidates[0] if external_candidates else None
missing_targets: list[str] = []
if tp1 is None:
missing_targets.append("missing_tp1_internal_target")
if tp2 is None:
missing_targets.append("missing_tp2_external_target")
status = TARGET_PLAN_STATUS_READY if not missing_targets else TARGET_PLAN_STATUS_INCOMPLETE
target_plan = TargetPlan(
status=status,
side=side,
tp1=tp1,
tp2=tp2,
management_rules=build_management_rules(trading_plan=trading_plan),
missing_targets=missing_targets,
candidates={
"internal": [item.to_dict() for item in internal_candidates],
"external": [item.to_dict() for item in external_candidates],
},
meta={
"entry_price": entry_price,
"range_id": range_row.get("id") if range_row else None,
"source_counts": {
"liquidity_pools": len(liquidity_pools),
"sessions": len(sessions),
"candles": len(candles),
},
},
)
return target_plan
def compute_entry_price(*, entry_low: float, entry_high: float) -> float:
entry_low = float(entry_low or 0.0)
entry_high = float(entry_high or 0.0)
if entry_low and entry_high:
return (entry_low + entry_high) / 2
return entry_low or entry_high
def build_internal_target_candidates(
*,
side: str,
entry_price: float,
range_row: Optional[dict],
liquidity_pools: list[dict],
sessions: list[dict],
candles: list[dict],
) -> list[TargetSelection]:
candidates: list[TargetSelection] = []
range_low = _float_or_none((range_row or {}).get("low"))
range_high = _float_or_none((range_row or {}).get("high"))
equilibrium = _float_or_none((range_row or {}).get("equilibrium"))
if equilibrium is not None and is_favorable_price(side=side, entry_price=entry_price, target_price=equilibrium):
candidates.append(
build_selection(
label="TP1",
layer=OBJECTIVE_LAYER_IRL,
price=equilibrium,
source_type="dealing_range_equilibrium",
table="dealing_ranges",
source_id=(range_row or {}).get("id"),
field="equilibrium",
reason="nearest_internal_range_liquidity",
)
)
for pool in liquidity_pools:
price = liquidity_target_price(side=side, pool=pool)
if price is None or not is_favorable_price(side=side, entry_price=entry_price, target_price=price):
continue
if is_inside_range(price=price, range_low=range_low, range_high=range_high):
candidates.append(
build_selection(
label="TP1",
layer=OBJECTIVE_LAYER_IRL,
price=price,
source_type="internal_liquidity_pool",
table="liquidity_pools",
source_id=pool.get("id"),
field=liquidity_target_field(side),
reason="internal_liquidity_before_external_expansion",
)
)
for session in sessions:
for field in session_internal_fields(side):
price = _float_or_none(session.get(field))
if price is None or not is_favorable_price(side=side, entry_price=entry_price, target_price=price):
continue
if is_inside_range(price=price, range_low=range_low, range_high=range_high):
candidates.append(
build_selection(
label="TP1",
layer=OBJECTIVE_LAYER_IRL,
price=price,
source_type="session_pivot",
table="market_sessions",
source_id=session.get("id"),
field=field,
reason="session_internal_pivot",
)
)
std_price = build_std_projection(side=side, entry_price=entry_price, candles=candles, multiplier=1.0)
if std_price is not None and is_inside_range(price=std_price, range_low=range_low, range_high=range_high):
candidates.append(
build_selection(
label="TP1",
layer=OBJECTIVE_LAYER_IRL,
price=std_price,
source_type="std_projection",
table="candles",
source_id=None,
field="close_std_1",
reason="one_standard_deviation_internal_projection",
)
)
return sort_target_candidates(side=side, entry_price=entry_price, candidates=dedupe_target_candidates(candidates))
def build_external_target_candidates(
*,
side: str,
entry_price: float,
range_row: Optional[dict],
liquidity_pools: list[dict],
sessions: list[dict],
candles: list[dict],
) -> list[TargetSelection]:
candidates: list[TargetSelection] = []
range_low = _float_or_none((range_row or {}).get("low"))
range_high = _float_or_none((range_row or {}).get("high"))
boundary_field = "high" if side == "buy" else "low"
boundary_price = _float_or_none((range_row or {}).get(boundary_field))
if boundary_price is not None and is_favorable_price(side=side, entry_price=entry_price, target_price=boundary_price):
candidates.append(
build_selection(
label="TP2",
layer=OBJECTIVE_LAYER_ERL,
price=boundary_price,
source_type="dealing_range_boundary",
table="dealing_ranges",
source_id=(range_row or {}).get("id"),
field=boundary_field,
reason="external_range_liquidity_boundary",
)
)
for pool in liquidity_pools:
price = liquidity_target_price(side=side, pool=pool)
if price is None or not is_favorable_price(side=side, entry_price=entry_price, target_price=price):
continue
if is_external_to_range(side=side, price=price, range_low=range_low, range_high=range_high):
candidates.append(
build_selection(
label="TP2",
layer=OBJECTIVE_LAYER_ERL,
price=price,
source_type="external_liquidity_pool",
table="liquidity_pools",
source_id=pool.get("id"),
field=liquidity_target_field(side),
reason="external_liquidity_after_internal_partial",
)
)
for session in sessions:
field = "high" if side == "buy" else "low"
price = _float_or_none(session.get(field))
if price is None or not is_favorable_price(side=side, entry_price=entry_price, target_price=price):
continue
if is_external_to_range(side=side, price=price, range_low=range_low, range_high=range_high):
candidates.append(
build_selection(
label="TP2",
layer=OBJECTIVE_LAYER_ERL,
price=price,
source_type="session_external_pivot",
table="market_sessions",
source_id=session.get("id"),
field=field,
reason="session_external_pivot",
)
)
adr_price = build_adr_projection(side=side, entry_price=entry_price, candles=candles)
if adr_price is not None and is_favorable_price(side=side, entry_price=entry_price, target_price=adr_price):
candidates.append(
build_selection(
label="TP2",
layer=OBJECTIVE_LAYER_ERL,
price=adr_price,
source_type="adr_projection",
table="candles",
source_id=None,
field="average_range",
reason="adr_external_projection",
)
)
std_price = build_std_projection(side=side, entry_price=entry_price, candles=candles, multiplier=2.0)
if std_price is not None and is_favorable_price(side=side, entry_price=entry_price, target_price=std_price):
candidates.append(
build_selection(
label="TP2",
layer=OBJECTIVE_LAYER_ERL,
price=std_price,
source_type="std_projection",
table="candles",
source_id=None,
field="close_std_2",
reason="two_standard_deviation_external_projection",
)
)
return sort_target_candidates(side=side, entry_price=entry_price, candidates=dedupe_target_candidates(candidates))
def build_selection(
*,
label: str,
layer: str,
price: float,
source_type: str,
table: str,
source_id,
field: str,
reason: str,
) -> TargetSelection:
return TargetSelection(
label=label,
objective_layer=layer,
price=float(price),
source={
"type": source_type,
"table": table,
"id": source_id,
"field": field,
},
selection_reason=reason,
)
def build_management_rules(trading_plan: Optional[dict] = None) -> list[dict]:
allowed_sessions = list((trading_plan or {}).get("allowed_sessions") or [])
invalidations = list((trading_plan or {}).get("invalidations") or [])
return [
{
"code": "tp1_partial",
"trigger": "tp1_hit",
"action": "partial_close",
"fraction": 0.5,
},
{
"code": "move_stop",
"trigger": "tp1_hit",
"action": "move_stop",
"to": "breakeven",
},
{
"code": "invalidate_on_reclaim",
"trigger": "invalidation_reclaim",
"action": "exit",
"invalidation_sources": invalidations,
},
{
"code": "expire_by_session",
"trigger": "session_end",
"action": "cancel_unfilled_or_flatten",
"allowed_sessions": allowed_sessions,
},
]
def has_complete_structural_target_plan(target_plan: Optional[dict]) -> bool:
if not isinstance(target_plan, dict):
return False
if target_plan.get("status") != TARGET_PLAN_STATUS_READY:
return False
tp1 = target_plan.get("tp1") or {}
tp2 = target_plan.get("tp2") or {}
return target_has_traceable_source(tp1, OBJECTIVE_LAYER_IRL) and target_has_traceable_source(tp2, OBJECTIVE_LAYER_ERL)
def target_has_traceable_source(target: dict, expected_layer: str) -> bool:
source = dict(target.get("source") or {})
return (
target.get("objective_layer") == expected_layer
and target.get("price") is not None
and bool(source.get("type"))
and bool(source.get("table"))
and bool(source.get("field"))
)
def liquidity_target_price(*, side: str, pool: dict) -> Optional[float]:
if side == "buy":
return _float_or_none(pool.get("price_high") if pool.get("price_high") is not None else pool.get("price_low"))
if side == "sell":
return _float_or_none(pool.get("price_low") if pool.get("price_low") is not None else pool.get("price_high"))
return None
def liquidity_target_field(side: str) -> str:
return "price_high" if side == "buy" else "price_low"
def session_internal_fields(side: str) -> tuple[str, ...]:
if side == "buy":
return ("midpoint", "high")
return ("midpoint", "low")
def is_favorable_price(*, side: str, entry_price: float, target_price: float) -> bool:
if side == "buy":
return target_price > entry_price
if side == "sell":
return target_price < entry_price
return False
def is_inside_range(*, price: float, range_low: Optional[float], range_high: Optional[float]) -> bool:
if range_low is None or range_high is None:
return True
return range_low <= price <= range_high
def is_external_to_range(*, side: str, price: float, range_low: Optional[float], range_high: Optional[float]) -> bool:
if range_low is None or range_high is None:
return True
if side == "buy":
return price >= range_high
if side == "sell":
return price <= range_low
return False
def build_adr_projection(*, side: str, entry_price: float, candles: list[dict]) -> Optional[float]:
ranges = [
abs(float(candle.get("high") or 0.0) - float(candle.get("low") or 0.0))
for candle in candles
if candle.get("high") is not None and candle.get("low") is not None
]
ranges = [value for value in ranges if value > 0]
if not ranges:
return None
projected = mean(ranges)
return entry_price + projected if side == "buy" else entry_price - projected
def build_std_projection(*, side: str, entry_price: float, candles: list[dict], multiplier: float) -> Optional[float]:
closes = [float(candle.get("close")) for candle in candles if candle.get("close") is not None]
if len(closes) < 2:
return None
projected = pstdev(closes) * multiplier
if projected <= 0:
return None
return entry_price + projected if side == "buy" else entry_price - projected
def sort_target_candidates(*, side: str, entry_price: float, candidates: list[TargetSelection]) -> list[TargetSelection]:
return sorted(
candidates,
key=lambda item: (
_target_source_priority(item),
abs(item.price - entry_price),
),
)
def dedupe_target_candidates(candidates: list[TargetSelection]) -> list[TargetSelection]:
seen = set()
result = []
for candidate in candidates:
key = (
candidate.objective_layer,
round(candidate.price, 8),
candidate.source.get("type"),
candidate.source.get("table"),
candidate.source.get("id"),
candidate.source.get("field"),
)
if key in seen:
continue
seen.add(key)
result.append(candidate)
return result
def _float_or_none(value) -> Optional[float]:
if value is None:
return None
return float(value)
def _target_source_priority(candidate: TargetSelection) -> int:
source_type = str(candidate.source.get("type") or "")
if candidate.objective_layer == OBJECTIVE_LAYER_IRL:
priority = {
"session_pivot": 0,
"dealing_range_equilibrium": 1,
"internal_liquidity_pool": 2,
"std_projection": 3,
}
return priority.get(source_type, 9)
priority = {
"dealing_range_boundary": 0,
"external_liquidity_pool": 1,
"session_external_pivot": 2,
"adr_projection": 3,
"std_projection": 4,
}
return priority.get(source_type, 9)
src/services/trade_plan_candidates/setup_watcher_service.py
+73 -5
View File
@@ -4,6 +4,7 @@ import math
from typing import Optional
from src.repositories.trade_plan_candidate_repository import TradePlanCandidateRepository
from src.services.targets import TargetSelectionService, has_complete_structural_target_plan
from src.services.trade_plan_candidates.confirmation_service import (
ConfirmationService,
ConfirmationState,
@@ -59,9 +60,11 @@ class SetupWatcherService:
self,
repository: Optional[TradePlanCandidateRepository] = None,
confirmation_service: Optional[ConfirmationService] = None,
target_selection_service: Optional[TargetSelectionService] = None,
) -> None:
self.repository = repository or TradePlanCandidateRepository()
self.confirmation_service = confirmation_service or ConfirmationService()
self.target_selection_service = target_selection_service or TargetSelectionService()
def build_candidate(
self,
@@ -86,6 +89,28 @@ class SetupWatcherService:
mss = self.repository.fetch_latest_mss(instrument_id=instrument_id, timeframe=setup_tf)
fvgs = self.repository.fetch_recent_fvgs(instrument_id=instrument_id, timeframe=trigger_tf, limit=20)
fvg = select_fvg_for_candidate(fvgs=fvgs, mss=mss, side=side)
range_row = self.repository.fetch_latest_range(instrument_id=instrument_id, timeframe=setup_tf)
liquidity_pools = self.repository.fetch_recent_liquidity_pools(
instrument_id=instrument_id,
timeframe=setup_tf,
limit=50,
)
sessions = self.repository.fetch_recent_sessions_for_targets(instrument_id=instrument_id, limit=10)
candles = self.repository.fetch_recent_candles_for_targets(
instrument_id=instrument_id,
timeframe=trigger_tf,
limit=100,
)
target_plan = self.target_selection_service.select_targets(
side=side or "unknown",
entry_low=float(fvg.get("lower")) if fvg and fvg.get("lower") is not None else 0.0,
entry_high=float(fvg.get("upper")) if fvg and fvg.get("upper") is not None else 0.0,
trading_plan=trading_plan,
range_row=range_row,
liquidity_pools=liquidity_pools,
sessions=sessions,
candles=candles,
)
confirmation = self.confirmation_service.evaluate(
trading_plan=trading_plan,
sweep=sweep,
@@ -101,6 +126,7 @@ class SetupWatcherService:
mss=mss,
fvg=fvg,
confirmation=confirmation,
target_plan=target_plan.to_dict(),
created_ts=created_ts,
)
@@ -125,6 +151,7 @@ def build_trade_plan_candidate_from_inputs(
mss: Optional[dict],
fvg: Optional[dict],
confirmation: ConfirmationState,
target_plan: Optional[dict] = None,
created_ts=None,
) -> TradePlanCandidateDraft:
created_ts = created_ts or datetime.now(timezone.utc)
@@ -135,14 +162,28 @@ def build_trade_plan_candidate_from_inputs(
evidence = {
**confirmation.evidence,
"primary_draw": trading_plan.get("primary_draw") or {},
"target_plan": target_plan or {},
"timeframe_set": {
"setup_tf": setup_tf,
"trigger_tf": trigger_tf,
},
}
entry = build_candidate_entry(side=side, mss=mss, fvg=fvg)
entry = build_candidate_entry(side=side, mss=mss, fvg=fvg, target_plan=target_plan)
invalidations = build_candidate_invalidations(trading_plan=trading_plan, sweep=sweep, mss=mss, fvg=fvg)
missing_confirmations = list(confirmation.missing_conditions)
reason_codes = list(confirmation.reason_codes)
rejected_conditions = list(confirmation.rejected_conditions)
if not has_complete_structural_target_plan(target_plan):
missing_confirmations.append("structural_targets_confirmed")
reason_codes.extend((target_plan or {}).get("missing_targets") or ["missing_structural_targets"])
status = "blocked"
missing_confirmations = _dedupe(missing_confirmations)
reason_codes = _dedupe(reason_codes)
rejected_conditions = _dedupe(rejected_conditions)
confirmation_state["status"] = status
confirmation_state["executable"] = confirmation.executable and has_complete_structural_target_plan(target_plan)
confirmation_state["missing_conditions"] = missing_confirmations
confirmation_state["reason_codes"] = reason_codes
return TradePlanCandidateDraft(
instrument_id=int(trading_plan["instrument_id"]),
trading_plan_id=int(trading_plan["id"]),
@@ -166,9 +207,11 @@ def build_trade_plan_candidate_from_inputs(
),
meta={
"source": "SetupWatcherService",
"executable": confirmation.executable,
"reason_codes": confirmation.reason_codes,
"rejected_conditions": confirmation.rejected_conditions,
"executable": confirmation.executable and has_complete_structural_target_plan(target_plan),
"reason_codes": reason_codes,
"rejected_conditions": rejected_conditions,
"target_plan": target_plan or {},
"management_rules": list((target_plan or {}).get("management_rules") or []),
},
)
@@ -192,7 +235,12 @@ def select_fvg_for_candidate(fvgs: list[dict], mss: Optional[dict], side: Option
return direction_matches[0]
def build_candidate_entry(side: Optional[str], mss: Optional[dict], fvg: Optional[dict]) -> dict:
def build_candidate_entry(
side: Optional[str],
mss: Optional[dict],
fvg: Optional[dict],
target_plan: Optional[dict] = None,
) -> dict:
if side not in {"buy", "sell"} or fvg is None:
return {}
entry_low = _float_or_none(fvg.get("lower"))
@@ -211,6 +259,15 @@ def build_candidate_entry(side: Optional[str], mss: Optional[dict], fvg: Optiona
entry["stop_loss"] = min(broken_level, entry_low) * 0.999
else:
entry["stop_loss"] = max(broken_level, entry_high) * 1.001
tp1 = (target_plan or {}).get("tp1") or {}
tp2 = (target_plan or {}).get("tp2") or {}
if tp1.get("price") is not None:
entry["tp1"] = float(tp1["price"])
if tp2.get("price") is not None:
entry["tp2"] = float(tp2["price"])
if target_plan:
entry["target_plan"] = target_plan
entry["management_rules"] = list(target_plan.get("management_rules") or [])
return entry
@@ -269,3 +326,14 @@ def _float_or_none(value) -> Optional[float]:
if math.isnan(parsed):
return None
return parsed
def _dedupe(values: list[str]) -> list[str]:
seen = set()
result = []
for value in values:
if value in seen:
continue
seen.add(value)
result.append(value)
return result
tests/unit/test_backtest_service.py
+11
View File
@@ -365,6 +365,15 @@ class BacktestServiceRunTests(unittest.TestCase):
"bias_tf": "1m",
"setup_tf": "1m",
"trigger_tf": "1m",
"signal_meta": {
"target_plan": {
"status": "ready",
"tp1": {"price": 102.0, "objective_layer": "IRL", "source": {"type": "session_pivot", "table": "market_sessions", "field": "high"}},
"tp2": {"price": 104.0, "objective_layer": "ERL", "source": {"type": "dealing_range_boundary", "table": "dealing_ranges", "field": "high"}},
"management_rules": [{"code": "tp1_partial"}],
},
"management_rules": [{"code": "tp1_partial"}],
},
}
]
@@ -395,6 +404,8 @@ class BacktestServiceRunTests(unittest.TestCase):
self.assertEqual(len(repository.results), 1)
self.assertEqual(repository.results[0]["meta"]["signal_symbol"], "BTC-USDT")
self.assertEqual(repository.results[0]["meta"]["signal_timeframe_bucket"], "1m/1m/1m")
self.assertEqual(repository.results[0]["meta"]["target_plan"]["tp1"]["objective_layer"], "IRL")
self.assertEqual(repository.results[0]["meta"]["management_rules"][0]["code"], "tp1_partial")
self.assertEqual(repository.summary["by_symbol"]["BTC-USDT"]["count"], 1)
self.assertEqual(repository.summary["by_timeframe"]["1m/1m/1m"]["count"], 1)
self.assertIn("quality_gate", repository.summary)
tests/unit/test_execution_service.py
+25
View File
@@ -176,6 +176,31 @@ class ExecutionDecisionTests(unittest.TestCase):
self.assertIn("execution_kill_switch_enabled", decision.reasons)
self.assertTrue(decision.meta["account_risk"]["kill_switch"])
def test_rejects_modeled_signal_without_structural_targets(self) -> None:
decision = evaluate_execution_decision(
signal={
"model_code": "SWEEP_MSS_FVG",
"side": "buy",
"status": "awaiting_confirmation",
"entry_type": "limit",
"entry_low": 100.0,
"entry_high": 101.0,
"stop_loss": 99.0,
"tp1": 102.0,
"tp2": 104.0,
"session_code": "LONDON",
"execution_intensity": "medium",
"meta": {"execution_permission": True, "target_plan": {}, "management_rules": []},
},
execution_mode="paper",
account_equity=10000.0,
risk_pct=0.01,
)
self.assertFalse(decision.allowed)
self.assertIn("missing_structural_targets", decision.reasons)
self.assertIn("missing_management_rules", decision.reasons)
class ExecutionServiceTests(unittest.TestCase):
def test_semi_auto_creation_requires_explicit_confirmation(self) -> None:
tests/unit/test_signal_service.py
+24
View File
@@ -192,6 +192,10 @@ class SignalRRFilterTests(unittest.TestCase):
self.assertEqual(candidate.meta["fvg_touch_count"], 0)
self.assertEqual(candidate.meta["fvg_data_quality_warnings"], [])
self.assertEqual(candidate.invalidations, ["mss_invalidated", "fvg_mitigated"])
self.assertEqual(candidate.meta["target_selection_status"], "ready")
self.assertEqual(candidate.meta["target_plan"]["tp1"]["objective_layer"], "IRL")
self.assertEqual(candidate.meta["target_plan"]["tp2"]["objective_layer"], "ERL")
self.assertTrue(candidate.meta["management_rules"])
def test_build_signal_rejects_candidate_when_required_evidence_is_missing(self) -> None:
service = SignalService(
@@ -222,6 +226,7 @@ class SignalRRFilterTests(unittest.TestCase):
self.assertEqual(candidate.evidence["liquidity_target_id"], 17)
self.assertEqual(candidate.evidence["displacement_id"], 7)
self.assertTrue(candidate.meta["evidence_quality"]["is_complete"])
self.assertEqual(candidate.meta["target_plan"]["tp1"]["source"]["table"], "market_sessions")
class SignalEvidenceTests(unittest.TestCase):
@@ -317,6 +322,25 @@ class _SignalStubRepository:
return None
return self.sweep
def fetch_latest_range(self, instrument_id: int, timeframe: str):
return {"id": 10, "low": 99.0, "high": 103.0, "equilibrium": 101.0, "state": "valid"}
def fetch_recent_liquidity_pools(self, instrument_id: int, timeframe: str, limit: int = 50):
return [
{"id": 17, "side": "buy_side", "price_low": 100.8, "price_high": 101.0, "status": "untouched"},
{"id": 18, "side": "buy_side", "price_low": 103.5, "price_high": 104.0, "status": "untouched"},
]
def fetch_recent_sessions_for_targets(self, instrument_id: int, limit: int = 10):
return [{"id": 11, "session_code": "LONDON", "high": 102.0, "low": 99.4, "midpoint": 100.6}]
def fetch_recent_candles_for_targets(self, instrument_id: int, timeframe: str, limit: int = 100):
return [
{"high": 101.0, "low": 100.0, "close": 100.5},
{"high": 102.0, "low": 100.5, "close": 101.2},
{"high": 102.8, "low": 101.0, "close": 102.2},
]
class SignalIdempotencyTests(unittest.TestCase):
def test_detects_equivalent_trade_signal(self) -> None:
tests/unit/test_target_selection_service.py
+66
View File
@@ -0,0 +1,66 @@
import os
import unittest
os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://user:pass@localhost:5432/ai_ict_test")
from src.services.targets import (
TARGET_PLAN_STATUS_INCOMPLETE,
TARGET_PLAN_STATUS_READY,
TargetSelectionService,
has_complete_structural_target_plan,
)
class TargetSelectionServiceTests(unittest.TestCase):
def test_selects_internal_tp1_and_external_tp2_with_traceable_sources(self) -> None:
plan = TargetSelectionService().select_targets(
side="buy",
entry_low=100.0,
entry_high=100.2,
trading_plan={"allowed_sessions": ["LONDON"], "invalidations": [{"code": "range_low_lost"}]},
range_row={"id": 10, "low": 99.0, "high": 103.0, "equilibrium": 101.0},
liquidity_pools=[
{"id": 21, "side": "buy_side", "price_low": 100.9, "price_high": 101.1, "status": "untouched"},
{"id": 22, "side": "buy_side", "price_low": 103.8, "price_high": 104.0, "status": "untouched"},
],
sessions=[{"id": 31, "session_code": "LONDON", "high": 102.0, "low": 99.2, "midpoint": 100.6}],
candles=[
{"high": 101.0, "low": 100.0, "close": 100.5},
{"high": 102.0, "low": 100.5, "close": 101.2},
{"high": 102.8, "low": 101.0, "close": 102.2},
],
)
payload = plan.to_dict()
self.assertEqual(payload["status"], TARGET_PLAN_STATUS_READY)
self.assertEqual(payload["tp1"]["objective_layer"], "IRL")
self.assertEqual(payload["tp2"]["objective_layer"], "ERL")
self.assertEqual(payload["tp1"]["source"]["table"], "market_sessions")
self.assertEqual(payload["tp2"]["source"]["table"], "dealing_ranges")
self.assertTrue(has_complete_structural_target_plan(payload))
self.assertEqual(
[rule["code"] for rule in payload["management_rules"]],
["tp1_partial", "move_stop", "invalidate_on_reclaim", "expire_by_session"],
)
def test_reports_missing_structural_targets_when_no_irl_or_erl_exists(self) -> None:
plan = TargetSelectionService().select_targets(
side="buy",
entry_low=100.0,
entry_high=100.2,
range_row=None,
liquidity_pools=[],
sessions=[],
candles=[],
)
payload = plan.to_dict()
self.assertEqual(payload["status"], TARGET_PLAN_STATUS_INCOMPLETE)
self.assertEqual(payload["missing_targets"], ["missing_tp1_internal_target", "missing_tp2_external_target"])
self.assertFalse(has_complete_structural_target_plan(payload))
if __name__ == "__main__":
unittest.main()
tests/unit/test_trade_plan_candidate_service.py
+22
View File
@@ -57,6 +57,9 @@ class SetupWatcherServiceTests(unittest.TestCase):
self.assertEqual(candidate.entry["entry_type"], "limit")
self.assertEqual(candidate.entry["entry_low"], 100.0)
self.assertEqual(candidate.entry["entry_high"], 100.2)
self.assertEqual(candidate.entry["target_plan"]["tp1"]["objective_layer"], "IRL")
self.assertEqual(candidate.entry["target_plan"]["tp2"]["objective_layer"], "ERL")
self.assertTrue(candidate.meta["management_rules"])
self.assertIn("active_trading_plan_invalidated", [item["code"] for item in candidate.invalidations])
def test_missing_sweep_mss_and_fvg_do_not_produce_executable_candidate(self) -> None:
@@ -211,6 +214,25 @@ class _SetupWatcherStubRepository:
def fetch_recent_fvgs(self, instrument_id: int, timeframe: str, limit: int = 20):
return self.fvgs
def fetch_latest_range(self, instrument_id: int, timeframe: str):
return {"id": 77, "low": 99.0, "high": 103.0, "equilibrium": 101.0, "state": "valid"}
def fetch_recent_liquidity_pools(self, instrument_id: int, timeframe: str, limit: int = 50):
return [
{"id": 21, "side": "buy_side", "price_low": 100.8, "price_high": 101.0, "status": "untouched"},
{"id": 22, "side": "buy_side", "price_low": 103.5, "price_high": 104.0, "status": "untouched"},
]
def fetch_recent_sessions_for_targets(self, instrument_id: int, limit: int = 10):
return [{"id": 11, "session_code": "LONDON", "high": 102.0, "low": 99.4, "midpoint": 100.6}]
def fetch_recent_candles_for_targets(self, instrument_id: int, timeframe: str, limit: int = 100):
return [
{"high": 101.0, "low": 100.0, "close": 100.5},
{"high": 102.0, "low": 100.5, "close": 101.2},
{"high": 102.8, "low": 101.0, "close": 102.2},
]
def fetch_latest_candidate(self, trading_plan_id: int, model_code: str):
return self.latest_candidate