mahmoud20138 Trading Brain -- 7-Layer Autonomous Trading Architecture
Role: Master orchestrator | Priority: Highest | Level: Advanced Supersedes
trading-brainas the primary entry point.
Architecture Overview
USER QUERY
|
v
+=========================================================================+
| TRADING BRAIN (Layer 0) |
| Parse intent -> Route -> Orchestrate -> Synthesize -> Respond |
+=========================================================================+
| | | | | |
v v v v v v
+----------+----------+----------+----------+----------+----------+
| LAYER 1 | LAYER 2 | LAYER 3 | LAYER 4 | LAYER 5 | LAYER 6 |---+
| Market | Strategy | Signal | Signal | Risk | Execution| |
| Intel | Select | Generate | Aggregate| Engine | Engine | |
+----------+----------+----------+----------+----------+----------+ |
^ |
| +----------+ |
+------------------------| LAYER 7 |<----------------------------+
| Learning |
| Engine |
+----------+
Detailed Layer Diagram
+=======================================================================+
| L1: MARKET INTELLIGENCE ENGINE |
| "What is the market doing right now?" |
| |
| +------------------+ +------------------+ +------------------+ |
| | market- | | macro-regime | | news-sentiment | |
| | intelligence | | market-regime- | | news-intelligence| |
| | fundamental- | | classifier | | economic- | |
| | analysis | | market-regime- | | calendar | |
| | | | detection | | sentiment- | |
| | | | | | analysis | |
| +------------------+ +------------------+ +------------------+ |
| +------------------+ +------------------+ +------------------+ |
| | mt5-chart- | | cross-asset- | | institutional- | |
| | browser | | relationships | | timeline | |
| | market-data- | | correlation- | | liquidity- | |
| | ingestion | | fundamentals | | analysis | |
| +------------------+ +------------------+ +------------------+ |
| |
| OUTPUT: MarketState { regime, sentiment, key_levels, volatility, |
| correlations, news_events, institutional_flow } |
+=======================================================================+
|
v
+=======================================================================+
| L2: STRATEGY SELECTION ENGINE |
| "Which strategies work in this market state?" |
| |
| +------------------+ +------------------+ +------------------+ |
| | strategy- | | portfolio- | | market-regime- | |
| | selection | | strategy- | | classifier | |
| | strategy- | | allocation | | multi-strategy- | |
| | playbook | | multi-strategy- | | allocator | |
| | | | orchestration | | | |
| +------------------+ +------------------+ +------------------+ |
| |
| OUTPUT: list[Strategy] with weights and regime-fit scores |
+=======================================================================+
|
v
+=======================================================================+
| L3: SIGNAL GENERATION |
| "Run selected strategies, produce raw signals" |
| |
| +------------------+ +------------------+ +------------------+ |
| | ict-smart-money | | trend-following- | | mean-reversion- | |
| | price-action | | systems | | engine | |
| | supply-demand- | | breakout- | | divergence- | |
| | zone-strategy | | strategy-engine | | strategy-engine | |
| | market-structure | | momentum-roc- | | fibonacci- | |
| | -bos-choch | | strategy | | strategy-engine | |
| +------------------+ +------------------+ +------------------+ |
| +------------------+ +------------------+ +------------------+ |
| | volume-profile- | | chart-vision | | session-breakout | |
| | strategy | | mtf-confluence- | | -strategies | |
| | wyckoff-method | | scorer | | scalping- | |
| | -engine | | technical- | | framework | |
| | | | analysis | | | |
| +------------------+ +------------------+ +------------------+ |
| |
| OUTPUT: list[Signal] { direction, entry, stop, targets, confidence } |
+=======================================================================+
|
v
+=======================================================================+
| L4: SIGNAL AGGREGATION |
| "Combine signals into a single decision" |
| |
| +------------------+ +------------------+ +------------------+ |
| | ai-signal- | | signal- | | mtf-confluence- | |
| | aggregator | | aggregator | | scorer | |
| +------------------+ +------------------+ +------------------+ |
| |
| OUTPUT: FinalSignal { direction, conviction, entry, stop, targets, |
| conflict_flags, agreement_score } |
+=======================================================================+
|
v
+=======================================================================+
| L5: RISK ENGINE |
| "Is this trade safe to take?" |
| |
| +------------------+ +------------------+ +------------------+ |
| | risk-and-portfolio | | risk-manager- | | drawdown- | |
| | risk-and- | | position-sizer | | playbook | |
| | portfolio | | risk-of-ruin | | drawdown- | |
| | | | | | recovery- | |
| | | | | | protocol | |
| +------------------+ +------------------+ +------------------+ |
| +------------------+ +------------------+ +------------------+ |
| | correlation- | | risk-calendar- | | tail-risk- | |
| | crisis | | trade-filter | | hedging | |
| | real-time-risk- | | risk-and-portfolio | | account-tail- | |
| | monitor | | | | risk | |
| +------------------+ +------------------+ +------------------+ |
| |
| OUTPUT: RiskApproval { approved, position_size, adjusted_stop, |
| max_loss, portfolio_heat, veto_reasons } |
+=======================================================================+
|
v
+=======================================================================+
| L6: EXECUTION ENGINE |
| "Place the trade optimally" |
| |
| +------------------+ +------------------+ +------------------+ |
| | execution-algo- | | execution- | | mt5-integration | |
| | trading | | optimizer | | mt5-ea-code- | |
| | market-impact- | | spread-slippage- | | generator | |
| | model | | cost-analyzer | | trading- | |
| | | | | | automation | |
| +------------------+ +------------------+ +------------------+ |
| +------------------+ +------------------+ |
| | signal- | | alert-pipeline | |
| | broadcaster | | realtime-alert- | |
| | | | pipeline | |
| +------------------+ +------------------+ |
| |
| OUTPUT: TradeResult { order_id, fill_price, slippage, status } |
+=======================================================================+
|
v
+=======================================================================+
| L7: LEARNING ENGINE |
| "What did we learn? How do we improve?" |
| |
| +------------------+ +------------------+ +------------------+ |
| | tensortrade-rl | | strategy- | | trade-journal- | |
| | rl-trade-agent | | validation | | analytics | |
| | strategy-decay- | | walk-forward- | | trade-journal- | |
| | monitor | | optimizer | | performance | |
| +------------------+ +------------------+ +------------------+ |
| +------------------+ +------------------+ +------------------+ |
| | performance- | | parameter- | | monte-carlo- | |
| | attribution- | | sensitivity- | | stress-tester | |
| | engine | | analyzer | | backtesting-sim | |
| +------------------+ +------------------+ +------------------+ |
| |
| OUTPUT: LearningUpdate { weight_adjustments, strategy_scores, |
| decay_alerts, performance_attribution } |
| FEEDBACK LOOP -> L1 (regime weights), L2 (strategy scores), |
| L4 (aggregation weights), L5 (risk params) |
+=======================================================================+
Layer Mapping Table
| Layer | Purpose | Super Skills | Micro Skills |
|---|---|---|---|
| L1 Market Intelligence | Understand current market state | market-intelligence, macro-regime, news-sentiment |
fundamental-analysis, social-sentiment-scraper, economic-calendar, news-intelligence, news-sentiment-nlp-engine, social-sentiment-scraper, institutional-timeline, market-regime-classifier, market-regime-classifier, mt5-chart-browser, market-data-ingestion, cross-asset-relationships, correlation-fundamentals, liquidity-analysis, currency-strength-meter, macro-economic-dashboard |
| L2 Strategy Selection | Pick optimal strategies for regime | strategy-selection, multi-strategy-orchestration, portfolio-optimization |
strategy-playbook, market-regime-classifier, multi-strategy-allocator, backtesting-sim, price-action |
| L3 Signal Generation | Run strategies, produce raw signals | technical-analysis, price-action, ict-smart-money |
trend-following-systems, mean-reversion-engine, breakout-strategy-engine, divergence-strategy-engine, fibonacci-strategy-engine, harmonic-pattern-engine, elliott-wave-engine, poc-bounce-strategy, market-structure-bos-choch, volume-profile-strategy, wyckoff-method-engine, momentum-roc-strategy, chart-vision, mtf-confluence-scorer, session-breakout-strategies, scalping-framework, order-flow-delta-strategy, smart-money-trap-detector, ichimoku-complete-strategy, candlestick-patterns |
| L4 Signal Aggregation | Combine signals into one decision | ai-signal-aggregator, ai-signal-aggregator |
mtf-confluence-scorer |
| L5 Risk Engine | Validate risk, size position | risk-and-portfolio, risk-and-portfolio |
risk-and-portfolio, risk-of-ruin, drawdown-playbook, drawdown-playbook, correlation-crisis, real-time-risk-monitor, risk-calendar-trade-filter, tail-risk-hedging, account-tail-risk, risk-and-portfolio, risk-adjusted-compounding, trade-psychology-coach |
| L6 Execution Engine | Place trades optimally | execution-algo-trading, mt5-integration, trading-autopilot |
execution-algo-trading, market-impact-model, spread-slippage-cost-analyzer, mt5-ea-code-generator, signal-broadcaster, alert-pipeline, realtime-alert-pipeline, trade-copier-signal-broadcaster |
| L7 Learning Engine | Adapt weights, detect decay, journal | tensortrade-rl, strategy-validation, trade-journal-analytics |
rl-trade-agent, strategy-decay-monitor, walk-forward-optimizer, parameter-sensitivity-analyzer, monte-carlo-stress-tester, backtesting-sim, performance-attribution-engine, trade-journal-performance, strategy-ab-tester, backtest-report-generator |
Data Flow Between Layers
L1 MarketState -----> L2 (regime, volatility, session, correlations)
L2 Strategies -----> L3 (strategy list with weights and parameters)
L3 RawSignals -----> L4 (direction, confidence, entry/stop/target per strategy)
L4 FinalSignal -----> L5 (aggregated direction, conviction, conflict flags)
L5 RiskApproval ----> L6 (position size, adjusted stops, go/no-go, veto reasons)
L6 TradeResult ----> L7 (fill data, slippage, execution quality)
L7 LearningUpdate --> L1 (updated regime weights)
--> L2 (updated strategy fitness scores)
--> L4 (updated aggregation weights)
--> L5 (updated risk parameters, drawdown state)
Data Contracts
from dataclasses import dataclass, field
from typing import Optional, Literal
from datetime import datetime
from enum import Enum
class Regime(Enum):
TRENDING_BULL = "trending_bull"
TRENDING_BEAR = "trending_bear"
RANGING = "ranging"
VOLATILE = "volatile"
TRANSITIONING = "transitioning"
class Direction(Enum):
LONG = "long"
SHORT = "short"
FLAT = "flat"
@dataclass
class MarketState:
"""L1 output: complete snapshot of current market conditions."""
symbol: str
timestamp: datetime
regime: Regime
regime_confidence: float # 0-1
trend_direction: Direction
volatility_percentile: float # 0-100, current ATR vs 252-day
adr_ratio: float # today ADR / 20-day avg ADR
key_levels: dict # {"support": [...], "resistance": [...]}
sentiment_score: float # -1 (extreme bearish) to +1 (extreme bullish)
news_events: list[dict] # upcoming high-impact events
institutional_bias: Direction
correlation_shifts: list[dict] # regime breaks in correlated pairs
session: str # "london", "new_york", "tokyo", "overlap"
currency_strength: dict[str, float] # {"USD": 0.7, "EUR": -0.3, ...}
@dataclass
class Strategy:
"""L2 output: a selected strategy with regime-fit metadata."""
name: str
skill_name: str # maps to a SKILL.md
weight: float # allocation weight 0-1
regime_fit: float # how well it fits current regime 0-1
parameters: dict # strategy-specific params
expected_win_rate: float
expected_rr: float
timeframes: list[str] # ["H1", "H4"]
@dataclass
class Signal:
"""L3 output: raw signal from a single strategy."""
strategy_name: str
direction: Direction
entry_price: float
stop_loss: float
targets: list[float]
confidence: float # 0-1
timeframe: str
reasoning: str
confluence_factors: list[str]
@dataclass
class FinalSignal:
"""L4 output: aggregated signal from all strategies."""
direction: Direction
conviction: float # 0-1, weighted agreement
entry_price: float
stop_loss: float
targets: list[float]
agreement_score: float # % of strategies agreeing
conflict_flags: list[str] # disagreements worth noting
contributing_signals: list[Signal]
veto: bool # True if conflicts are irreconcilable
@dataclass
class RiskApproval:
"""L5 output: risk validation result."""
approved: bool
position_size_lots: float
position_size_pct: float # of account
adjusted_stop: Optional[float] # risk engine may widen stop
max_loss_dollars: float
max_loss_pct: float
portfolio_heat_pct: float # total open risk
risk_reward_ratio: float
veto_reasons: list[str] # why rejected (if not approved)
conditions: list[str] # "reduce size 50%", "set trailing stop"
@dataclass
class TradeResult:
"""L6 output: execution result."""
order_id: str
symbol: str
direction: Direction
fill_price: float
requested_price: float
slippage_pips: float
spread_at_fill: float
execution_algo: str # "market", "limit", "twap", etc.
status: Literal["filled", "partial", "rejected", "pending"]
timestamp: datetime
@dataclass
class LearningUpdate:
"""L7 output: feedback for all layers."""
strategy_scores: dict[str, float] # updated fitness per strategy
weight_adjustments: dict[str, float] # delta to aggregation weights
decay_alerts: list[str] # strategies losing edge
risk_param_updates: dict[str, float] # adjusted risk thresholds
regime_weight_updates: dict[str, float] # regime detection calibration
performance_summary: dict # Sharpe, win rate, avg RR, etc.
TradingBrain Implementation
"""
TradingBrain -- Master orchestrator for the 7-layer trading architecture.
Integrates with skill_router.py for skill discovery and routing.
Each layer reads the relevant SKILL.md files and executes their patterns.
Usage:
from trading_brain import TradingBrain
brain = TradingBrain(symbol="EURUSD", account_equity=10000)
result = brain.run()
"""
from __future__ import annotations
import json
import os
import sys
from dataclasses import asdict
from datetime import datetime
from pathlib import Path
from typing import Optional
# Import skill router for dynamic skill lookup
SKILLS_DIR = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(SKILLS_DIR))
from skill_router import load_skills, route_query, resolve_deps, find_skill
# ---------------------------------------------------------------------------
# Layer -> Skill mapping
# ---------------------------------------------------------------------------
LAYER_SKILLS: dict[str, list[str]] = {
"L1_intelligence": [
"market-intelligence", "macro-regime", "news-sentiment",
"market-regime-classifier", "market-regime-classifier",
"mt5-chart-browser", "cross-asset-relationships",
"correlation-fundamentals", "institutional-timeline",
"liquidity-analysis", "social-sentiment-scraper",
"economic-calendar", "currency-strength-meter"],
"L2_strategy": [
"strategy-selection", "strategy-playbook",
"portfolio-optimization", "multi-strategy-orchestration",
"multi-strategy-allocator", "market-regime-classifier"],
"L3_signals": [
"ict-smart-money", "price-action", "technical-analysis",
"trend-following-systems", "mean-reversion-engine",
"breakout-strategy-engine", "divergence-strategy-engine",
"poc-bounce-strategy", "market-structure-bos-choch",
"volume-profile-strategy", "wyckoff-method-engine",
"fibonacci-strategy-engine", "momentum-roc-strategy",
"chart-vision", "mtf-confluence-scorer",
"session-breakout-strategies", "scalping-framework"],
"L4_aggregation": [
"ai-signal-aggregator", "ai-signal-aggregator",
"mtf-confluence-scorer"],
"L5_risk": [
"risk-and-portfolio", "risk-and-portfolio",
"risk-and-portfolio", "risk-of-ruin",
"drawdown-playbook", "drawdown-playbook",
"correlation-crisis", "real-time-risk-monitor",
"risk-calendar-trade-filter", "tail-risk-hedging"],
"L6_execution": [
"execution-algo-trading", "execution-algo-trading",
"mt5-integration", "mt5-ea-code-generator",
"trading-autopilot", "signal-broadcaster",
"alert-pipeline", "spread-slippage-cost-analyzer"],
"L7_learning": [
"tensortrade-rl", "strategy-decay-monitor",
"rl-trade-agent", "strategy-validation",
"walk-forward-optimizer", "trade-journal-analytics",
"performance-attribution-engine", "monte-carlo-stress-tester",
"backtesting-sim", "parameter-sensitivity-analyzer"],
}
class TradingBrain:
"""
Master orchestrator implementing the 7-layer autonomous trading pipeline.
Each method corresponds to one layer. The run() method executes the
full pipeline sequentially, with each layer feeding the next.
"""
def __init__(
self,
symbol: str = "EURUSD",
timeframes: list[str] = None,
account_equity: float = 10_000.0,
risk_per_trade_pct: float = 1.0,
max_portfolio_heat_pct: float = 6.0,
mode: str = "analysis", # "analysis" | "live" | "paper"
):
self.symbol = symbol
self.timeframes = timeframes or ["M15", "H1", "H4", "D1"]
self.account_equity = account_equity
self.risk_per_trade_pct = risk_per_trade_pct
self.max_portfolio_heat_pct = max_portfolio_heat_pct
self.mode = mode
# State flowing between layers
self.market_state: Optional[MarketState] = None
self.strategies: list[Strategy] = []
self.signals: list[Signal] = []
self.final_signal: Optional[FinalSignal] = None
self.risk_approval: Optional[RiskApproval] = None
self.trade_result: Optional[TradeResult] = None
self.learning_update: Optional[LearningUpdate] = None
# Skill registry
self._skills = load_skills()
# Execution log
self.log: list[dict] = []
# -------------------------------------------------------------------
# Layer 1: Market Intelligence
# -------------------------------------------------------------------
def analyze_market(self) -> MarketState:
"""
Layer 1: Gather all market intelligence for self.symbol.
Reads skills: market-intelligence, macro-regime, news-sentiment,
mt5-chart-browser, cross-asset-relationships, institutional-timeline.
Returns MarketState with regime, sentiment, levels, correlations.
"""
self._log("L1_INTEL", f"Analyzing market for {self.symbol}")
# Step 1: Get price data and compute indicators (mt5-chart-browser)
# Step 2: Classify regime (market-regime-classifier + market-regime-classifier)
# Step 3: Fetch news and sentiment (news-sentiment, economic-calendar)
# Step 4: Check institutional positioning (institutional-timeline)
# Step 5: Run correlation analysis (cross-asset-relationships)
# Step 6: Calculate currency strength (currency-strength-meter)
skills_to_invoke = self._resolve_layer_skills("L1_intelligence")
self._log("L1_INTEL", f"Invoking {len(skills_to_invoke)} skills")
# Build MarketState from aggregated skill outputs
self.market_state = MarketState(
symbol=self.symbol,
timestamp=datetime.utcnow(),
regime=Regime.RANGING, # populated by L1 skills
regime_confidence=0.0,
trend_direction=Direction.FLAT,
volatility_percentile=50.0,
adr_ratio=1.0,
key_levels={"support": [], "resistance": []},
sentiment_score=0.0,
news_events=[],
institutional_bias=Direction.FLAT,
correlation_shifts=[],
session=self._detect_session(),
currency_strength={},
)
self._log("L1_INTEL", f"Regime={self.market_state.regime.value}, "
f"Session={self.market_state.session}")
return self.market_state
# -------------------------------------------------------------------
# Layer 2: Strategy Selection
# -------------------------------------------------------------------
def select_strategy(self, market_state: MarketState) -> list[Strategy]:
"""
Layer 2: Select optimal strategies for the current market state.
Reads skills: strategy-selection, strategy-playbook,
portfolio-optimization, multi-strategy-orchestration.
Returns ranked list of Strategy objects with weights.
"""
self._log("L2_STRATEGY", f"Selecting strategies for regime={market_state.regime.value}")
# Strategy-regime mapping
REGIME_STRATEGY_MAP = {
Regime.TRENDING_BULL: [
("trend-following-systems", 0.35),
("ict-smart-money", 0.25),
("breakout-strategy-engine", 0.20),
("momentum-roc-strategy", 0.20)],
Regime.TRENDING_BEAR: [
("trend-following-systems", 0.35),
("ict-smart-money", 0.25),
("breakout-strategy-engine", 0.20),
("momentum-roc-strategy", 0.20)],
Regime.RANGING: [
("mean-reversion-engine", 0.30),
("poc-bounce-strategy", 0.25),
("divergence-strategy-engine", 0.25),
("scalping-framework", 0.20)],
Regime.VOLATILE: [
("breakout-strategy-engine", 0.30),
("session-breakout-strategies", 0.25),
("volume-profile-strategy", 0.25),
("fibonacci-strategy-engine", 0.20)],
Regime.TRANSITIONING: [
("market-structure-bos-choch", 0.30),
("wyckoff-method-engine", 0.25),
("smart-money-trap-detector", 0.25),
("divergence-strategy-engine", 0.20)],
}
regime_strategies = REGIME_STRATEGY_MAP.get(
market_state.regime, REGIME_STRATEGY_MAP[Regime.RANGING]
)
self.strategies = []
for skill_name, weight in regime_strategies:
self.strategies.append(Strategy(
name=skill_name.replace("-", " ").title(),
skill_name=skill_name,
weight=weight,
regime_fit=market_state.regime_confidence,
parameters={},
expected_win_rate=0.0, # populated by L7 feedback
expected_rr=0.0,
timeframes=self.timeframes,
))
self._log("L2_STRATEGY", f"Selected {len(self.strategies)} strategies: "
f"{[s.skill_name for s in self.strategies]}")
return self.strategies
# -------------------------------------------------------------------
# Layer 3: Signal Generation
# -------------------------------------------------------------------
def generate_signals(self, strategies: list[Strategy]) -> list[Signal]:
"""
Layer 3: Run each selected strategy to produce raw signals.
For each Strategy, reads its SKILL.md and applies its analysis
to the current market data. Returns one Signal per strategy.
"""
self._log("L3_SIGNALS", f"Generating signals from {len(strategies)} strategies")
self.signals = []
for strategy in strategies:
skill_path = self._get_skill_path(strategy.skill_name)
if not skill_path:
self._log("L3_SIGNALS", f"SKIP: {strategy.skill_name} not found")
continue
# Each strategy skill produces a Signal with:
# - direction, entry, stop, targets, confidence, reasoning
signal = Signal(
strategy_name=strategy.skill_name,
direction=Direction.FLAT, # populated by strategy execution
entry_price=0.0,
stop_loss=0.0,
targets=[],
confidence=0.0,
timeframe=strategy.timeframes[0] if strategy.timeframes else "H1",
reasoning="",
confluence_factors=[],
)
self.signals.append(signal)
self._log("L3_SIGNALS", f"Generated {len(self.signals)} raw signals")
return self.signals
# -------------------------------------------------------------------
# Layer 4: Signal Aggregation
# -------------------------------------------------------------------
def aggregate_signals(self, signals: list[Signal]) -> FinalSignal:
"""
Layer 4: Combine all raw signals into a single FinalSignal.
Uses weighted voting from ai-signal-aggregator and conflict
detection. Produces conviction score and identifies disagreements.
"""
self._log("L4_AGGREGATE", f"Aggregating {len(signals)} signals")
if not signals:
self.final_signal = FinalSignal(
direction=Direction.FLAT, conviction=0.0,
entry_price=0.0, stop_loss=0.0, targets=[],
agreement_score=0.0, conflict_flags=["NO_SIGNALS"],
contributing_signals=[], veto=True,
)
return self.final_signal
# Weighted vote
long_score = 0.0
short_score = 0.0
total_weight = 0.0
for sig in signals:
# Look up strategy weight from L2
weight = 1.0
for strat in self.strategies:
if strat.skill_name == sig.strategy_name:
weight = strat.weight
break
effective_weight = weight * sig.confidence
if sig.direction == Direction.LONG:
long_score += effective_weight
elif sig.direction == Direction.SHORT:
short_score += effective_weight
total_weight += effective_weight
# Determine direction and conviction
if total_weight == 0:
direction = Direction.FLAT
conviction = 0.0
elif long_score > short_score:
direction = Direction.LONG
conviction = min((long_score - short_score) / total_weight, 0.95)
elif short_score > long_score:
direction = Direction.SHORT
conviction = min((short_score - long_score) / total_weight, 0.95)
else:
direction = Direction.FLAT
conviction = 0.0
# Conflict detection
directions = [s.direction for s in signals if s.direction != Direction.FLAT]
long_count = sum(1 for d in directions if d == Direction.LONG)
short_count = sum(1 for d in directions if d == Direction.SHORT)
total_directional = long_count + short_count
conflict_flags = []
if long_count > 0 and short_count > 0:
conflict_flags.append(
f"CONFLICT: {long_count} LONG vs {short_count} SHORT signals"
)
agreement = max(long_count, short_count) / max(total_directional, 1)
# Veto if agreement below threshold
veto = agreement < 0.5 or conviction < 0.15
self.final_signal = FinalSignal(
direction=direction,
conviction=round(conviction, 3),
entry_price=0.0, # best entry from contributing signals
stop_loss=0.0, # tightest stop from contributing signals
targets=[],
agreement_score=round(agreement, 3),
conflict_flags=conflict_flags,
contributing_signals=signals,
veto=veto,
)
self._log("L4_AGGREGATE",
f"Direction={direction.value}, Conviction={conviction:.3f}, "
f"Agreement={agreement:.1%}, Veto={veto}")
return self.final_signal
# -------------------------------------------------------------------
# Layer 5: Risk Engine
# -------------------------------------------------------------------
def validate_risk(self, signal: FinalSignal) -> RiskApproval:
"""
Layer 5: Validate the trade against all risk rules.
Checks: position sizing, portfolio heat, drawdown state,
correlation risk, event calendar, risk-of-ruin, and tail risk.
Returns RiskApproval with go/no-go decision.
"""
self._log("L5_RISK", "Validating risk")
veto_reasons = []
# Rule 1: Signal must not be vetoed by L4
if signal.veto:
veto_reasons.append("L4 aggregation vetoed: insufficient agreement")
# Rule 2: Conviction floor
if signal.conviction < 0.30:
veto_reasons.append(f"Conviction {signal.conviction:.1%} below 30% floor")
# Rule 3: Conflict flags
if signal.conflict_flags:
veto_reasons.append(f"Unresolved conflicts: {signal.conflict_flags}")
# Rule 4: Portfolio heat check
current_heat = 0.0 # would query real-time-risk-monitor
remaining_heat = self.max_portfolio_heat_pct - current_heat
if remaining_heat <= 0:
veto_reasons.append(f"Portfolio heat {current_heat}% exceeds max {self.max_portfolio_heat_pct}%")
# Rule 5: Event calendar filter (risk-calendar-trade-filter)
# Would check if high-impact news is within blackout window
# Rule 6: Drawdown state (drawdown-playbook)
# Would check current drawdown and apply size reduction rules
# Position sizing (risk-and-portfolio)
if signal.stop_loss and signal.entry_price:
stop_distance = abs(signal.entry_price - signal.stop_loss)
dollar_risk = self.account_equity * (self.risk_per_trade_pct / 100)
position_size = dollar_risk / stop_distance if stop_distance > 0 else 0
rr_ratio = 0.0
if signal.targets and stop_distance > 0:
target_distance = abs(signal.targets[0] - signal.entry_price)
rr_ratio = target_distance / stop_distance
# Rule 7: Minimum R:R
if rr_ratio < 1.5:
veto_reasons.append(f"R:R {rr_ratio:.1f} below 1.5 minimum")
else:
dollar_risk = self.account_equity * (self.risk_per_trade_pct / 100)
position_size = 0.0
rr_ratio = 0.0
approved = len(veto_reasons) == 0
# Apply conviction-based size adjustment
if approved and signal.conviction < 0.60:
position_size *= 0.5 # half size for moderate conviction
conditions = ["Reduced to 50% size (conviction < 60%)"]
else:
conditions = []
self.risk_approval = RiskApproval(
approved=approved,
position_size_lots=round(position_size, 2),
position_size_pct=round(self.risk_per_trade_pct * (0.5 if signal.conviction < 0.60 else 1.0), 2),
adjusted_stop=signal.stop_loss,
max_loss_dollars=round(dollar_risk, 2),
max_loss_pct=self.risk_per_trade_pct,
portfolio_heat_pct=current_heat + self.risk_per_trade_pct,
risk_reward_ratio=round(rr_ratio, 2),
veto_reasons=veto_reasons,
conditions=conditions,
)
self._log("L5_RISK", f"Approved={approved}, Size={position_size:.2f}, "
f"Veto reasons={veto_reasons}")
return self.risk_approval
# -------------------------------------------------------------------
# Layer 6: Execution Engine
# -------------------------------------------------------------------
def execute(self, signal: FinalSignal, risk: RiskApproval) -> TradeResult:
"""
Layer 6: Execute the trade if risk-approved.
Uses execution-algo-trading for optimal entry, mt5-integration
for order placement, and alert-pipeline for notifications.
"""
self._log("L6_EXEC", f"Execution mode={self.mode}")
if not risk.approved:
self.trade_result = TradeResult(
order_id="REJECTED",
symbol=self.symbol,
direction=signal.direction,
fill_price=0.0,
requested_price=signal.entry_price,
slippage_pips=0.0,
spread_at_fill=0.0,
execution_algo="none",
status="rejected",
timestamp=datetime.utcnow(),
)
self._log("L6_EXEC", f"REJECTED: {risk.veto_reasons}")
return self.trade_result
if self.mode == "analysis":
# Analysis-only mode: report what WOULD be traded
self.trade_result = TradeResult(
order_id="ANALYSIS_ONLY",
symbol=self.symbol,
direction=signal.direction,
fill_price=signal.entry_price,
requested_price=signal.entry_price,
slippage_pips=0.0,
spread_at_fill=0.0,
execution_algo="analysis",
status="pending",
timestamp=datetime.utcnow(),
)
self._log("L6_EXEC", "Analysis mode: trade plan generated, not executed")
elif self.mode == "paper":
# Paper trading via trade-simulator-paper
self.trade_result = TradeResult(
order_id=f"PAPER_{datetime.utcnow().strftime('%Y%m%d%H%M%S')}",
symbol=self.symbol,
direction=signal.direction,
fill_price=signal.entry_price,
requested_price=signal.entry_price,
slippage_pips=0.0,
spread_at_fill=0.0,
execution_algo="paper",
status="filled",
timestamp=datetime.utcnow(),
)
self._log("L6_EXEC", "Paper trade logged")
else:
# Live mode: use execution-algo-trading + mt5-integration
# Would invoke MT5 order placement here
self.trade_result = TradeResult(
order_id=f"LIVE_{datetime.utcnow().strftime('%Y%m%d%H%M%S')}",
symbol=self.symbol,
…(truncated)