Add broker reconciliation trade loop

.env.example

@@ -42,6 +42,7 @@ TRADING_RUNTIME_SETUP_TIMEFRAME=1m
 TRADING_RUNTIME_TRIGGER_TIMEFRAME=1m
 TRADING_RUNTIME_MARKET_STATE_INTERVAL_SECONDS=15
 TRADING_RUNTIME_BROKER_SYNC_INTERVAL_SECONDS=30
+TRADING_RUNTIME_RECONCILIATION_INTERVAL_SECONDS=45
 TRADING_RUNTIME_WATCH_INTERVAL_SECONDS=20
 TRADING_RUNTIME_DECISION_PACKET_INTERVAL_SECONDS=20
 TRADING_RUNTIME_PLAN_INTERVAL_SECONDS=900

migrations/031_add_real_trade_attribution_fields.sql

@@ -0,0 +1,8 @@
+alter table trading_journal_entries
+    add column if not exists real_trade_detected boolean not null default false,
+    add column if not exists out_of_model_trade boolean not null default false,
+    add column if not exists reconciliation_status text,
+    add column if not exists broker_fill_count integer not null default 0;
+
+create index if not exists idx_trading_journal_entries_real_trade_date_desc
+on trading_journal_entries (real_trade_detected, journal_date desc, journal_ts desc);

scripts/run_broker_reconciliation.py

@@ -0,0 +1,37 @@
+from __future__ import annotations
+
+from pathlib import Path
+import json
+import os
+import sys
+
+ROOT = Path(__file__).resolve().parents[1]
+if str(ROOT) not in sys.path:
+    sys.path.insert(0, str(ROOT))
+
+from src.services.reconciliation import BrokerExecutionReconciliationService
+
+
+def main(env=None) -> int:
+    env = os.environ if env is None else env
+    instrument_id = int(env.get("BROKER_RECONCILIATION_INSTRUMENT_ID", "1"))
+    output_json = _get_bool(env, "BROKER_RECONCILIATION_OUTPUT_JSON", False)
+    summary = BrokerExecutionReconciliationService().reconcile_instrument(instrument_id=instrument_id)
+    if output_json:
+        print(json.dumps(summary, ensure_ascii=False, sort_keys=True))
+    else:
+        print(f"Reconciliation status: {summary.get('status')}")
+        print(f"Matched executions: {summary.get('matched_execution_ids') or []}")
+        print(f"Imported executions: {summary.get('imported_execution_ids') or []}")
+    return 0 if summary.get("status") == "ok" else 2
+
+
+def _get_bool(env, key: str, default: bool = False) -> bool:
+    value = env.get(key)
+    if value is None or value == "":
+        return default
+    return str(value).strip().lower() in {"1", "true", "yes", "on"}
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

src/api/__init__.py

@@ -31,6 +31,7 @@ from src.api.trading_runtime import (
     get_weekly_review,
     rerun_ticket_risk_check,
     run_candidate_confirm_check,
+    run_trading_reconciliation,
     run_trading_runtime_once,
     run_trading_watch,
     sync_execution,
@@ -67,6 +68,7 @@ __all__ = [
     "run_trading_watch",
     "get_trading_candidate_by_id",
     "run_candidate_confirm_check",
+    "run_trading_reconciliation",
     "run_trading_runtime_once",
     "build_ticket_from_candidate",
     "rerun_ticket_risk_check",

src/api/server.py

@@ -33,6 +33,7 @@ from src.api.trading_runtime import (
     get_weekly_review,
     rerun_ticket_risk_check,
     run_candidate_confirm_check,
+    run_trading_reconciliation,
     run_trading_runtime_once,
     run_trading_watch,
     sync_execution,
@@ -73,6 +74,7 @@ def build_api_dependencies() -> dict:
         "trading_market_state": get_trading_market_state,
         "trading_decision_packet": get_trading_decision_packet,
         "trading_decision_alerts": get_trading_decision_alerts,
+        "trading_reconciliation_run": run_trading_reconciliation,
         "trading_runtime_status": get_trading_runtime_status,
         "trading_runtime_run_once": run_trading_runtime_once,
         "trading_watch_run": run_trading_watch,
@@ -186,6 +188,9 @@ def route_request(
         instrument_id = _require_int(query, "instrument_id")
         limit = _get_int(query, "limit", 20)
         return 200, {"items": deps["trading_decision_alerts"](instrument_id=instrument_id, limit=limit)}
+    if method == "POST" and path == "/trading/reconcile":
+        instrument_id = _require_json_int(body or {}, "instrument_id")
+        return 200, deps["trading_reconciliation_run"](instrument_id=instrument_id)
     if method == "POST" and path == "/trading/watch/run":
         instrument_id = _require_json_int(body or {}, "instrument_id")
         candidate = deps["trading_watch_run"](

src/api/trading_runtime.py

@@ -12,6 +12,7 @@ from src.services.decision import DecisionPacketService
 from src.services.execution import ExecutionService
 from src.services.journal import TradingJournalService, WeeklyReviewService
 from src.services.market_state import MarketStateService, get_market_truth_supervisor
+from src.services.reconciliation import BrokerExecutionReconciliationService
 from src.services.runtime import get_trading_runtime_service
 from src.services.trade_plan_candidates import SetupWatcherService
 from src.services.trading_plans import ACTIVE_PLAN_STATUS, TradingPlanService
@@ -415,6 +416,10 @@ def get_trading_decision_alerts(instrument_id: int, limit: int = 20) -> list[dic
     return DecisionPacketService().fetch_recent_alerts(instrument_id=instrument_id, limit=limit)
 
 
+def run_trading_reconciliation(instrument_id: int) -> dict:
+    return BrokerExecutionReconciliationService().reconcile_instrument(instrument_id=instrument_id)
+
+
 def market_truth_degraded(market_state) -> bool:
     payload = market_state.to_dict() if hasattr(market_state, "to_dict") else dict(market_state or {})
     freshness_code = str(((payload.get("freshness") or {}).get("code")) or "")

src/config/settings.py

@@ -46,6 +46,7 @@ class Settings(BaseSettings):
     trading_runtime_trigger_timeframe: str = "1m"
     trading_runtime_market_state_interval_seconds: int = 15
     trading_runtime_broker_sync_interval_seconds: int = 30
+    trading_runtime_reconciliation_interval_seconds: int = 45
     trading_runtime_watch_interval_seconds: int = 20
     trading_runtime_decision_packet_interval_seconds: int = 20
     trading_runtime_plan_interval_seconds: int = 900

src/repositories/execution_repository.py

@@ -191,6 +191,42 @@ class ExecutionRepository:
             rows = session.execute(sql, {"instrument_id": instrument_id, "limit": limit}).mappings().all()
         return [dict(row) for row in rows]
 
+    def fetch_recent_execution_reconciliation_rows(self, instrument_id: int, limit: int = 200) -> list[dict]:
+        sql = text(
+            """
+            select
+                te.id as execution_id,
+                te.signal_id,
+                te.execution_ticket_id,
+                te.venue_order_id,
+                te.execution_mode,
+                te.status as execution_status,
+                te.side,
+                te.entry_price,
+                te.size,
+                te.opened_ts,
+                te.closed_ts,
+                te.broker_order_ids,
+                te.fill_state,
+                te.position_state,
+                te.close_reason,
+                te.reconciliation_state,
+                te.meta as execution_meta,
+                ts.instrument_id,
+                ts.model_code,
+                ts.session_code,
+                ts.meta as signal_meta
+            from trade_executions te
+            join trade_signals ts on ts.id = te.signal_id
+            where ts.instrument_id = :instrument_id
+            order by coalesce(te.closed_ts, te.opened_ts) desc nulls last, te.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 update_trade_execution_lifecycle(
         self,
         *,

src/repositories/signal_repository.py

@@ -347,3 +347,130 @@ class SignalRepository:
                     "meta": json.dumps(meta),
                 },
             )
+
+    def insert_trade_signal_returning_id(
+        self,
+        instrument_id: int,
+        model_code: str,
+        side: str,
+        status: str,
+        trading_plan_id: Optional[int],
+        candidate_id: Optional[int],
+        bias_snapshot_id,
+        entry_type: str,
+        entry_low: float,
+        entry_high: float,
+        stop_loss: float,
+        tp1: float,
+        tp2: float,
+        rr_tp1: float,
+        rr_tp2: float,
+        session_code,
+        created_ts,
+        expires_ts,
+        invalidations: list,
+        evidence: dict,
+        primary_draw: dict,
+        target_plan: dict,
+        risk_governor_decision: dict,
+        confirmation_quality: dict,
+        quality_score: Optional[float],
+        ai_explanation: Optional[str],
+        meta: dict,
+    ) -> int:
+        sql = text(
+            """
+            insert into trade_signals (
+                instrument_id,
+                model_code,
+                side,
+                status,
+                trading_plan_id,
+                candidate_id,
+                bias_snapshot_id,
+                entry_type,
+                entry_low,
+                entry_high,
+                stop_loss,
+                tp1,
+                tp2,
+                rr_tp1,
+                rr_tp2,
+                session_code,
+                created_ts,
+                expires_ts,
+                invalidations,
+                evidence,
+                primary_draw,
+                target_plan,
+                risk_governor_decision,
+                confirmation_quality,
+                quality_score,
+                ai_explanation,
+                meta
+            ) values (
+                :instrument_id,
+                :model_code,
+                :side,
+                :status,
+                :trading_plan_id,
+                :candidate_id,
+                :bias_snapshot_id,
+                :entry_type,
+                :entry_low,
+                :entry_high,
+                :stop_loss,
+                :tp1,
+                :tp2,
+                :rr_tp1,
+                :rr_tp2,
+                :session_code,
+                :created_ts,
+                :expires_ts,
+                cast(:invalidations as jsonb),
+                cast(:evidence as jsonb),
+                cast(:primary_draw as jsonb),
+                cast(:target_plan as jsonb),
+                cast(:risk_governor_decision as jsonb),
+                cast(:confirmation_quality as jsonb),
+                :quality_score,
+                :ai_explanation,
+                cast(:meta as jsonb)
+            )
+            returning id
+            """
+        )
+        with SessionLocal.begin() as session:
+            result = session.execute(
+                sql,
+                {
+                    "instrument_id": instrument_id,
+                    "model_code": model_code,
+                    "side": side,
+                    "status": status,
+                    "trading_plan_id": trading_plan_id,
+                    "candidate_id": candidate_id,
+                    "bias_snapshot_id": bias_snapshot_id,
+                    "entry_type": entry_type,
+                    "entry_low": entry_low,
+                    "entry_high": entry_high,
+                    "stop_loss": stop_loss,
+                    "tp1": tp1,
+                    "tp2": tp2,
+                    "rr_tp1": rr_tp1,
+                    "rr_tp2": rr_tp2,
+                    "session_code": session_code,
+                    "created_ts": created_ts,
+                    "expires_ts": expires_ts,
+                    "invalidations": json.dumps(invalidations, default=str),
+                    "evidence": json.dumps(evidence, default=str),
+                    "primary_draw": json.dumps(primary_draw, default=str),
+                    "target_plan": json.dumps(target_plan, default=str),
+                    "risk_governor_decision": json.dumps(risk_governor_decision, default=str),
+                    "confirmation_quality": json.dumps(confirmation_quality, default=str),
+                    "quality_score": quality_score,
+                    "ai_explanation": ai_explanation,
+                    "meta": json.dumps(meta, default=str),
+                },
+            )
+            return int(result.scalar_one())

src/repositories/trading_journal_repository.py

@@ -26,6 +26,11 @@ class TradingJournalRepository:
                 te.opened_ts,
                 te.closed_ts,
                 te.pnl,
+                te.broker_order_ids,
+                te.fill_state,
+                te.position_state,
+                te.close_reason,
+                te.reconciliation_state,
                 te.meta as execution_meta,
                 ts.instrument_id,
                 ts.model_code,
@@ -109,6 +114,10 @@ class TradingJournalRepository:
         framework_execution: bool,
         model_compliance: bool,
         risk_compliance: bool,
+        real_trade_detected: bool,
+        out_of_model_trade: bool,
+        reconciliation_status: Optional[str],
+        broker_fill_count: int,
         model_failure: bool,
         execution_violation: bool,
         risk_violation: bool,
@@ -131,6 +140,10 @@ class TradingJournalRepository:
                 framework_execution,
                 model_compliance,
                 risk_compliance,
+                real_trade_detected,
+                out_of_model_trade,
+                reconciliation_status,
+                broker_fill_count,
                 model_failure,
                 execution_violation,
                 risk_violation,
@@ -150,6 +163,10 @@ class TradingJournalRepository:
                 :framework_execution,
                 :model_compliance,
                 :risk_compliance,
+                :real_trade_detected,
+                :out_of_model_trade,
+                :reconciliation_status,
+                :broker_fill_count,
                 :model_failure,
                 :execution_violation,
                 :risk_violation,
@@ -170,6 +187,10 @@ class TradingJournalRepository:
                 framework_execution = excluded.framework_execution,
                 model_compliance = excluded.model_compliance,
                 risk_compliance = excluded.risk_compliance,
+                real_trade_detected = excluded.real_trade_detected,
+                out_of_model_trade = excluded.out_of_model_trade,
+                reconciliation_status = excluded.reconciliation_status,
+                broker_fill_count = excluded.broker_fill_count,
                 model_failure = excluded.model_failure,
                 execution_violation = excluded.execution_violation,
                 risk_violation = excluded.risk_violation,
@@ -196,6 +217,10 @@ class TradingJournalRepository:
                     "framework_execution": framework_execution,
                     "model_compliance": model_compliance,
                     "risk_compliance": risk_compliance,
+                    "real_trade_detected": real_trade_detected,
+                    "out_of_model_trade": out_of_model_trade,
+                    "reconciliation_status": reconciliation_status,
+                    "broker_fill_count": broker_fill_count,
                     "model_failure": model_failure,
                     "execution_violation": execution_violation,
                     "risk_violation": risk_violation,

src/services/journal/trading_journal_service.py

@@ -41,6 +41,10 @@ class TradingJournalEntryCandidate:
     framework_execution: bool
     model_compliance: bool
     risk_compliance: bool
+    real_trade_detected: bool
+    out_of_model_trade: bool
+    reconciliation_status: Optional[str]
+    broker_fill_count: int
     model_failure: bool
     execution_violation: bool
     risk_violation: bool
@@ -77,6 +81,12 @@ class TradingJournalService:
         execution_audit = dict(review.get("execution_audit") or {})
         error_types = list(review.get("error_types") or [])
         lifecycle_state = review.get("backtest_lifecycle_state") or classify_backtest_lifecycle_state_from_result(backtest_result)
+        execution_reconciliation = dict(trade_execution.get("reconciliation_state") or {})
+        real_trade_detected = is_real_trade_detected(execution_reconciliation)
+        out_of_model_trade = real_trade_detected and (
+            bool(execution_reconciliation.get("imported_external_trade"))
+            or str(trade_signal.get("meta", {}).get("structure_state") or "") == "external_trade"
+        )
 
         risk_violation = any(code in RISK_VIOLATION_CODES for code in error_types)
         execution_violation = execution_audit.get("audit_status") == "review" and not risk_violation and any(
@@ -89,7 +99,7 @@ class TradingJournalService:
             and not execution_violation
             and not plan_violation
         )
-        model_compliance = not plan_violation and _signal_structure_is_aligned(trade_signal)
+        model_compliance = not out_of_model_trade and not plan_violation and _signal_structure_is_aligned(trade_signal)
         risk_compliance = not risk_violation
         framework_execution = model_compliance and risk_compliance and not execution_violation and not model_failure
 
@@ -97,6 +107,8 @@ class TradingJournalService:
             trade_execution=trade_execution,
             backtest_result=backtest_result,
             framework_execution=framework_execution,
+            real_trade_detected=real_trade_detected,
+            out_of_model_trade=out_of_model_trade,
             risk_violation=risk_violation,
             execution_violation=execution_violation,
             plan_violation=plan_violation,
@@ -105,6 +117,9 @@ class TradingJournalService:
         next_actions = build_next_actions(
             review=review,
             lifecycle_state=lifecycle_state,
+            real_trade_detected=real_trade_detected,
+            out_of_model_trade=out_of_model_trade,
+            reconciliation_status=str(execution_reconciliation.get("status") or ""),
             risk_violation=risk_violation,
             execution_violation=execution_violation,
             plan_violation=plan_violation,
@@ -117,6 +132,7 @@ class TradingJournalService:
             "signal": trade_signal,
             "backtest_result": backtest_result,
             "ticket": _build_ticket_payload(bundle),
+            "real_trade_mapping": execution_reconciliation,
         }
         return TradingJournalEntryCandidate(
             execution_id=int(bundle["execution_id"]),
@@ -130,6 +146,10 @@ class TradingJournalService:
             framework_execution=framework_execution,
             model_compliance=model_compliance,
             risk_compliance=risk_compliance,
+            real_trade_detected=real_trade_detected,
+            out_of_model_trade=out_of_model_trade,
+            reconciliation_status=str(execution_reconciliation.get("status") or "") or None,
+            broker_fill_count=int(execution_reconciliation.get("fill_count") or len(execution_reconciliation.get("broker_fill_ids") or [])),
             model_failure=model_failure,
             execution_violation=execution_violation,
             risk_violation=risk_violation,
@@ -163,6 +183,8 @@ def build_review_tags(
     trade_execution: dict,
     backtest_result: Optional[dict],
     framework_execution: bool,
+    real_trade_detected: bool,
+    out_of_model_trade: bool,
     risk_violation: bool,
     execution_violation: bool,
     plan_violation: bool,
@@ -171,6 +193,10 @@ def build_review_tags(
     tags: list[str] = []
     if framework_execution:
         tags.append("framework_execution")
+    if real_trade_detected:
+        tags.append("real_trade_detected")
+    if out_of_model_trade:
+        tags.append("out_of_model_trade")
     if not risk_violation:
         tags.append("risk_compliant")
     if not execution_violation:
@@ -196,6 +222,9 @@ def build_next_actions(
     *,
     review: dict,
     lifecycle_state: Optional[str],
+    real_trade_detected: bool,
+    out_of_model_trade: bool,
+    reconciliation_status: str,
     risk_violation: bool,
     execution_violation: bool,
     plan_violation: bool,
@@ -208,6 +237,10 @@ def build_next_actions(
         actions.append("review_execution_process_and_confirmation_path")
     if plan_violation:
         actions.append("tighten_plan_and_candidate_validation")
+    if real_trade_detected and reconciliation_status in {"unmatched", "position_only", "imported_external_fill"}:
+        actions.append("review_real_trade_reconciliation_rules")
+    if out_of_model_trade:
+        actions.append("review_out_of_model_real_trades")
     if model_failure:
         actions.append("review_model_context_and_target_selection")
     if lifecycle_state == "tp2_closed":
@@ -241,6 +274,11 @@ def _build_trade_execution_payload(bundle: dict) -> dict:
         "status": bundle.get("execution_status"),
         "opened_ts": bundle.get("opened_ts"),
         "closed_ts": bundle.get("closed_ts"),
+        "broker_order_ids": list(bundle.get("broker_order_ids") or []),
+        "fill_state": bundle.get("fill_state"),
+        "position_state": bundle.get("position_state"),
+        "close_reason": bundle.get("close_reason"),
+        "reconciliation_state": dict(bundle.get("reconciliation_state") or {}),
         "meta": dict(bundle.get("execution_meta") or {}),
     }
 
@@ -300,3 +338,10 @@ def _dedupe(values: list[str]) -> list[str]:
         seen.add(text)
         result.append(text)
     return result
+
+
+def is_real_trade_detected(reconciliation_state: dict) -> bool:
+    status = str(reconciliation_state.get("status") or "")
+    if status in {"matched_fill", "matched_fill_flat", "position_only", "imported_external_fill"}:
+        return True
+    return bool(reconciliation_state.get("broker_fill_ids"))

src/services/journal/weekly_review_service.py

@@ -27,6 +27,11 @@ class WeeklyReviewService:
         framework_count = sum(1 for entry in entries if entry.get("framework_execution"))
         model_outside_count = sum(1 for entry in entries if not entry.get("model_compliance"))
         risk_violation_count = sum(1 for entry in entries if entry.get("risk_violation"))
+        real_trade_count = sum(1 for entry in entries if entry.get("real_trade_detected"))
+        out_of_model_real_trade_count = sum(1 for entry in entries if entry.get("out_of_model_trade"))
+        reconciliation_issue_count = sum(
+            1 for entry in entries if str(entry.get("reconciliation_status") or "") in {"unmatched", "position_only", "imported_external_fill"}
+        )
         session_stats = build_session_stats(entries)
         high_quality_session, low_quality_session = rank_sessions(session_stats)
         next_actions = recommend_next_actions(entries, low_quality_session=low_quality_session)
@@ -34,6 +39,9 @@ class WeeklyReviewService:
             "week_start": week_start,
             "week_end": week_end,
             "entry_count": total,
+            "real_trade_count": real_trade_count,
+            "out_of_model_real_trade_rate": _ratio(out_of_model_real_trade_count, real_trade_count),
+            "reconciliation_issue_count": reconciliation_issue_count,
             "framework_execution_rate": _ratio(framework_count, total),
             "out_of_model_trade_rate": _ratio(model_outside_count, total),
             "risk_violation_count": risk_violation_count,
@@ -51,6 +59,7 @@ def build_session_stats(entries: list[dict]) -> dict:
             "framework_count": 0,
             "risk_violations": 0,
             "model_outside": 0,
+            "real_trade_count": 0,
             "total_r": 0.0,
         }
     )
@@ -64,6 +73,8 @@ def build_session_stats(entries: list[dict]) -> dict:
             bucket["risk_violations"] += 1
         if not entry.get("model_compliance"):
             bucket["model_outside"] += 1
+        if entry.get("real_trade_detected"):
+            bucket["real_trade_count"] += 1
         summary = dict(entry.get("summary") or {})
         backtest_result = dict(summary.get("backtest_result") or {})
         backtest_meta = dict(backtest_result.get("meta") or {})
@@ -75,6 +86,7 @@ def build_session_stats(entries: list[dict]) -> dict:
             "framework_execution_rate": _ratio(bucket["framework_count"], bucket["count"]),
             "model_outside_rate": _ratio(bucket["model_outside"], bucket["count"]),
             "risk_violation_count": bucket["risk_violations"],
+            "real_trade_count": bucket["real_trade_count"],
             "avg_r_multiple": bucket["total_r"] / bucket["count"] if bucket["count"] else 0.0,
         }
     return stats
@@ -108,6 +120,10 @@ def recommend_next_actions(entries: list[dict], *, low_quality_session: Optional
     actions = [item for item, _count in counter.most_common(3)]
     if low_quality_session is not None:
         actions.append(f"review_{str(low_quality_session['session_code']).lower()}_session_playbook")
+    if any(entry.get("out_of_model_trade") for entry in entries):
+        actions.append("review_out_of_model_real_trades")
+    if any(str(entry.get("reconciliation_status") or "") in {"unmatched", "position_only", "imported_external_fill"} for entry in entries):
+        actions.append("tighten_real_trade_reconciliation")
     if not actions:
         actions = ["collect_more_executed_samples", "review_weekly_trade_selection", "review_risk_process"]
     deduped = []

src/services/reconciliation/__init__.py

@@ -0,0 +1,3 @@
+from src.services.reconciliation.broker_execution_reconciliation_service import BrokerExecutionReconciliationService
+
+__all__ = ["BrokerExecutionReconciliationService"]

src/services/reconciliation/broker_execution_reconciliation_service.py

@@ -0,0 +1,447 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from datetime import datetime, timedelta, timezone
+from typing import Optional
+
+from src.repositories.execution_repository import ExecutionRepository
+from src.repositories.signal_repository import SignalRepository
+from src.services.broker import BrokerReadModelService, BROKER_STATE_OK
+from src.services.journal import TradingJournalService
+
+
+EXTERNAL_REAL_TRADE_MODEL_CODE = "REAL_TRADE_IMPORT"
+REAL_TRADE_SIGNAL_STATUS = "external_fill_imported"
+
+
+@dataclass(frozen=True)
+class BrokerFillGroup:
+    group_key: str
+    order_id: str
+    instrument_code: str
+    side: str
+    fill_ids: list[str]
+    total_size: float
+    avg_price: float
+    first_fill_ts: Optional[datetime]
+    last_fill_ts: Optional[datetime]
+    raw_fills: list[dict]
+
+    def to_dict(self) -> dict:
+        return {
+            "group_key": self.group_key,
+            "order_id": self.order_id,
+            "instrument_code": self.instrument_code,
+            "side": self.side,
+            "fill_ids": list(self.fill_ids),
+            "total_size": self.total_size,
+            "avg_price": self.avg_price,
+            "first_fill_ts": self.first_fill_ts.isoformat() if self.first_fill_ts else None,
+            "last_fill_ts": self.last_fill_ts.isoformat() if self.last_fill_ts else None,
+            "raw_fills": list(self.raw_fills),
+        }
+
+
+class BrokerExecutionReconciliationService:
+    def __init__(
+        self,
+        *,
+        execution_repository: Optional[ExecutionRepository] = None,
+        signal_repository: Optional[SignalRepository] = None,
+        broker_read_model_service: Optional[BrokerReadModelService] = None,
+        journal_service: Optional[TradingJournalService] = None,
+    ) -> None:
+        self.execution_repository = execution_repository or ExecutionRepository()
+        self.signal_repository = signal_repository or SignalRepository()
+        self.broker_read_model_service = broker_read_model_service or BrokerReadModelService()
+        self.journal_service = journal_service or TradingJournalService()
+
+    def reconcile_instrument(
+        self,
+        *,
+        instrument_id: int,
+        now: Optional[datetime] = None,
+        limit: int = 200,
+    ) -> dict:
+        observed_now = _ensure_aware(now or datetime.now(timezone.utc))
+        snapshot = self.broker_read_model_service.fetch_snapshot(
+            instrument_id=instrument_id,
+            now=observed_now,
+            known_execution_refs={"broker_order_ids": [], "open_execution_count": 0},
+        )
+        if snapshot.status != BROKER_STATE_OK:
+            return {
+                "status": snapshot.status,
+                "error": snapshot.error,
+                "matched_execution_ids": [],
+                "imported_execution_ids": [],
+                "journal_updates": [],
+                "real_trade_groups": [],
+            }
+
+        execution_rows = self.execution_repository.fetch_recent_execution_reconciliation_rows(instrument_id=instrument_id, limit=limit)
+        fill_groups = build_fill_groups(snapshot.fills)
+        matched_execution_ids: list[int] = []
+        imported_execution_ids: list[int] = []
+        journal_updates: list[int] = []
+        claimed_group_keys: set[str] = set()
+        known_fill_ids = collect_known_fill_ids(execution_rows)
+        positions = list(snapshot.positions or [])
+        orders = list(snapshot.orders or [])
+
+        for row in execution_rows:
+            match = find_best_fill_group(row=row, fill_groups=fill_groups, claimed_group_keys=claimed_group_keys)
+            if match is None:
+                continue
+            claimed_group_keys.add(match.group_key)
+            updated_execution_id = self._apply_match(
+                row=row,
+                fill_group=match,
+                positions=positions,
+                orders=orders,
+                observed_now=observed_now,
+            )
+            matched_execution_ids.append(updated_execution_id)
+            journal_id = self.journal_service.upsert_entry_for_execution(updated_execution_id)
+            if journal_id:
+                journal_updates.append(journal_id)
+
+        for fill_group in fill_groups:
+            if fill_group.group_key in claimed_group_keys:
+                continue
+            if any(fill_id in known_fill_ids for fill_id in fill_group.fill_ids):
+                continue
+            execution_id = self._import_external_fill_group(
+                instrument_id=instrument_id,
+                fill_group=fill_group,
+                positions=positions,
+                orders=orders,
+                observed_now=observed_now,
+            )
+            imported_execution_ids.append(execution_id)
+            journal_id = self.journal_service.upsert_entry_for_execution(execution_id)
+            if journal_id:
+                journal_updates.append(journal_id)
+
+        return {
+            "status": "ok",
+            "matched_execution_ids": matched_execution_ids,
+            "imported_execution_ids": imported_execution_ids,
+            "journal_updates": journal_updates,
+            "real_trade_groups": [group.to_dict() for group in fill_groups],
+            "fills_seen": len(snapshot.fills or []),
+            "positions_seen": len(snapshot.positions or []),
+            "orders_seen": len(snapshot.orders or []),
+        }
+
+    def _apply_match(
+        self,
+        *,
+        row: dict,
+        fill_group: BrokerFillGroup,
+        positions: list[dict],
+        orders: list[dict],
+        observed_now: datetime,
+    ) -> int:
+        execution_id = int(row["execution_id"])
+        trade_execution = self.execution_repository.fetch_execution_by_id(execution_id) or {}
+        existing_meta = dict(trade_execution.get("meta") or {})
+        existing_reconciliation = dict(trade_execution.get("reconciliation_state") or existing_meta.get("reconciliation_state") or {})
+        broker_order_ids = sorted(set(list(trade_execution.get("broker_order_ids") or existing_meta.get("broker_order_ids") or []) + ([fill_group.order_id] if fill_group.order_id else [])))
+        position_detected = has_matching_position(fill_group=fill_group, positions=positions)
+        pending_order_detected = has_matching_order(fill_group=fill_group, orders=orders)
+        reconciliation_state = {
+            **existing_reconciliation,
+            "status": "matched_fill" if position_detected or pending_order_detected else "matched_fill_flat",
+            "matched_at": observed_now.isoformat(),
+            "broker_fill_ids": list(fill_group.fill_ids),
+            "broker_order_ids": broker_order_ids,
+            "fill_count": len(fill_group.fill_ids),
+            "fill_side": fill_group.side,
+            "avg_fill_price": fill_group.avg_price,
+            "total_fill_size": fill_group.total_size,
+            "position_detected": position_detected,
+            "pending_order_detected": pending_order_detected,
+            "imported_external_trade": False,
+            "confidence": "high" if fill_group.order_id else "medium",
+        }
+        meta = {
+            **existing_meta,
+            "broker_order_ids": broker_order_ids,
+            "reconciliation_state": reconciliation_state,
+            "real_trade_mapping": {
+                "fill_group": fill_group.to_dict(),
+                "matched_execution_id": execution_id,
+                "position_detected": position_detected,
+                "pending_order_detected": pending_order_detected,
+            },
+        }
+        status = trade_execution.get("status") or row.get("execution_status")
+        fill_state = trade_execution.get("fill_state") or row.get("fill_state") or "filled"
+        position_state = trade_execution.get("position_state") or row.get("position_state") or ("open" if position_detected else "closed")
+        close_reason = trade_execution.get("close_reason") or row.get("close_reason")
+        if status == "pending":
+            status = "open" if position_detected or pending_order_detected else "closed"
+        self.execution_repository.update_trade_execution_lifecycle(
+            execution_id=execution_id,
+            status=str(status),
+            closed_ts=trade_execution.get("closed_ts") or (observed_now if status == "closed" and not position_detected else None),
+            fill_state=fill_state,
+            position_state=position_state,
+            close_reason=close_reason,
+            reconciliation_state=reconciliation_state,
+            bracket_state=dict(trade_execution.get("bracket_state") or existing_meta.get("bracket_state") or {}),
+            broker_order_ids=broker_order_ids,
+            confirmed_by=trade_execution.get("confirmed_by"),
+            confirmed_at=trade_execution.get("confirmed_at"),
+            execution_ticket_id=trade_execution.get("execution_ticket_id"),
+            confirmation_required=bool(trade_execution.get("confirmation_required")),
+            quality_score=trade_execution.get("quality_score"),
+            ai_explanation=trade_execution.get("ai_explanation"),
+            meta=meta,
+        )
+        return execution_id
+
+    def _import_external_fill_group(
+        self,
+        *,
+        instrument_id: int,
+        fill_group: BrokerFillGroup,
+        positions: list[dict],
+        orders: list[dict],
+        observed_now: datetime,
+    ) -> int:
+        session_row = self.signal_repository.fetch_session_at(instrument_id=instrument_id, ts=fill_group.first_fill_ts or observed_now)
+        signal_id = self.signal_repository.insert_trade_signal_returning_id(
+            instrument_id=instrument_id,
+            model_code=EXTERNAL_REAL_TRADE_MODEL_CODE,
+            side=fill_group.side,
+            status=REAL_TRADE_SIGNAL_STATUS,
+            trading_plan_id=None,
+            candidate_id=None,
+            bias_snapshot_id=None,
+            entry_type="market",
+            entry_low=fill_group.avg_price,
+            entry_high=fill_group.avg_price,
+            stop_loss=0.0,
+            tp1=0.0,
+            tp2=0.0,
+            rr_tp1=0.0,
+            rr_tp2=0.0,
+            session_code=session_row.get("session_code") if session_row else None,
+            created_ts=fill_group.first_fill_ts or observed_now,
+            expires_ts=fill_group.last_fill_ts or observed_now,
+            invalidations=[],
+            evidence={
+                "external_real_trade": True,
+                "broker_order_id": fill_group.order_id,
+                "broker_fill_ids": list(fill_group.fill_ids),
+            },
+            primary_draw={},
+            target_plan={},
+            risk_governor_decision={},
+            confirmation_quality={},
+            quality_score=0.0,
+            ai_explanation=f"external_real_trade_import:{fill_group.instrument_code}:{fill_group.side}",
+            meta={
+                "structure_state": "external_trade",
+                "execution_permission": False,
+                "imported_real_trade": True,
+                "broker_fill_group": fill_group.to_dict(),
+            },
+        )
+        position_detected = has_matching_position(fill_group=fill_group, positions=positions)
+        pending_order_detected = has_matching_order(fill_group=fill_group, orders=orders)
+        reconciliation_state = {
+            "status": "imported_external_fill",
+            "matched_at": observed_now.isoformat(),
+            "broker_fill_ids": list(fill_group.fill_ids),
+            "broker_order_ids": [fill_group.order_id] if fill_group.order_id else [],
+            "fill_count": len(fill_group.fill_ids),
+            "fill_side": fill_group.side,
+            "avg_fill_price": fill_group.avg_price,
+            "total_fill_size": fill_group.total_size,
+            "position_detected": position_detected,
+            "pending_order_detected": pending_order_detected,
+            "imported_external_trade": True,
+            "confidence": "high",
+        }
+        execution_id = self.execution_repository.insert_trade_execution(
+            signal_id=signal_id,
+            execution_ticket_id=None,
+            venue_order_id=fill_group.order_id or fill_group.group_key,
+            execution_mode="manual",
+            side=fill_group.side,
+            entry_price=fill_group.avg_price,
+            stop_loss=0.0,
+            take_profit_1=0.0,
+            take_profit_2=0.0,
+            size=fill_group.total_size,
+            risk_pct=0.0,
+            status="open" if position_detected or pending_order_detected else "closed",
+            confirmation_required=False,
+            confirmed_by="broker_import",
+            confirmed_at=observed_now,
+            opened_ts=fill_group.first_fill_ts or observed_now,
+            broker_order_ids=[fill_group.order_id] if fill_group.order_id else [],
+            bracket_state={"status": "external_trade", "oco_group_id": None},
+            fill_state="filled",
+            position_state="open" if position_detected or pending_order_detected else "closed",
+            close_reason=None if position_detected or pending_order_detected else "external_trade_flat",
+            reconciliation_state=reconciliation_state,
+            quality_score=0.0,
+            ai_explanation=f"external_real_trade_import:{fill_group.side}",
+            meta={
+                "imported_real_trade": True,
+                "execution_source": "broker_fill_import",
+                "broker_fill_group": fill_group.to_dict(),
+                "reconciliation_state": reconciliation_state,
+                "structure_state": "external_trade",
+            },
+        )
+        if not position_detected and not pending_order_detected:
+            self.execution_repository.update_trade_execution_lifecycle(
+                execution_id=execution_id,
+                status="closed",
+                closed_ts=fill_group.last_fill_ts or observed_now,
+                fill_state="filled",
+                position_state="closed",
+                close_reason="external_trade_flat",
+                reconciliation_state=reconciliation_state,
+                bracket_state={"status": "external_trade", "oco_group_id": None},
+                broker_order_ids=[fill_group.order_id] if fill_group.order_id else [],
+                confirmed_by="broker_import",
+                confirmed_at=observed_now,
+                execution_ticket_id=None,
+                confirmation_required=False,
+                quality_score=0.0,
+                ai_explanation=f"external_real_trade_import:{fill_group.side}",
+                meta={
+                    "imported_real_trade": True,
+                    "execution_source": "broker_fill_import",
+                    "broker_fill_group": fill_group.to_dict(),
+                    "reconciliation_state": reconciliation_state,
+                    "structure_state": "external_trade",
+                },
+            )
+        return execution_id
+
+
+def build_fill_groups(fills: list[dict]) -> list[BrokerFillGroup]:
+    groups: dict[str, list[dict]] = {}
+    for fill in fills or []:
+        order_id = str(fill.get("order_id") or "").strip()
+        fill_id = str(fill.get("fill_id") or "").strip()
+        key = order_id or fill_id
+        if not key:
+            continue
+        groups.setdefault(key, []).append(dict(fill))
+    result = []
+    for key, rows in groups.items():
+        first = rows[0]
+        total_size = sum(float(row.get("fill_size") or 0.0) for row in rows)
+        total_notional = sum(float(row.get("fill_size") or 0.0) * float(row.get("fill_price") or 0.0) for row in rows)
+        avg_price = total_notional / total_size if total_size else float(first.get("fill_price") or 0.0)
+        fill_times = sorted((_parse_dt(row.get("fill_ts")) for row in rows if row.get("fill_ts")), key=lambda value: value or datetime.now(timezone.utc))
+        result.append(
+            BrokerFillGroup(
+                group_key=key,
+                order_id=str(first.get("order_id") or key),
+                instrument_code=str(first.get("instrument_id") or ""),
+                side=str(first.get("side") or "").lower(),
+                fill_ids=[str(row.get("fill_id") or "") for row in rows if row.get("fill_id")],
+                total_size=total_size,
+                avg_price=avg_price,
+                first_fill_ts=fill_times[0] if fill_times else None,
+                last_fill_ts=fill_times[-1] if fill_times else None,
+                raw_fills=rows,
+            )
+        )
+    return sorted(result, key=lambda item: item.last_fill_ts or datetime.min.replace(tzinfo=timezone.utc), reverse=True)
+
+
+def collect_known_fill_ids(execution_rows: list[dict]) -> set[str]:
+    known = set()
+    for row in execution_rows:
+        reconciliation_state = dict(row.get("reconciliation_state") or {})
+        for fill_id in list(reconciliation_state.get("broker_fill_ids") or []):
+            text = str(fill_id or "").strip()
+            if text:
+                known.add(text)
+    return known
+
+
+def find_best_fill_group(*, row: dict, fill_groups: list[BrokerFillGroup], claimed_group_keys: set[str]) -> Optional[BrokerFillGroup]:
+    broker_order_ids = {str(value).strip() for value in list(row.get("broker_order_ids") or []) if str(value or "").strip()}
+    venue_order_id = str(row.get("venue_order_id") or "").strip()
+    if venue_order_id:
+        broker_order_ids.add(venue_order_id)
+    for fill_group in fill_groups:
+        if fill_group.group_key in claimed_group_keys:
+            continue
+        if fill_group.order_id and fill_group.order_id in broker_order_ids:
+            return fill_group
+
+    execution_side = str(row.get("side") or "").lower()
+    execution_price = float(row.get("entry_price") or 0.0)
+    opened_ts = _parse_dt(row.get("opened_ts"))
+    if opened_ts is None or execution_price <= 0 or not execution_side:
+        return None
+    window_start = opened_ts - timedelta(hours=4)
+    window_end = opened_ts + timedelta(hours=4)
+    candidates = []
+    for fill_group in fill_groups:
+        if fill_group.group_key in claimed_group_keys:
+            continue
+        if fill_group.side != execution_side:
+            continue
+        if fill_group.first_fill_ts is None:
+            continue
+        if not (window_start <= fill_group.first_fill_ts <= window_end):
+            continue
+        if not price_is_close(fill_group.avg_price, execution_price):
+            continue
+        time_distance = abs((fill_group.first_fill_ts - opened_ts).total_seconds())
+        price_distance = abs(fill_group.avg_price - execution_price)
+        candidates.append((time_distance, price_distance, fill_group))
+    if not candidates:
+        return None
+    candidates.sort(key=lambda item: (item[0], item[1]))
+    return candidates[0][2]
+
+
+def has_matching_position(*, fill_group: BrokerFillGroup, positions: list[dict]) -> bool:
+    for position in positions or []:
+        instrument_code = str(position.get("instrument_id") or "")
+        if instrument_code != fill_group.instrument_code:
+            continue
+        side = str(position.get("side") or "").lower()
+        if side in {fill_group.side, "net"}:
+            return True
+    return False
+
+
+def has_matching_order(*, fill_group: BrokerFillGroup, orders: list[dict]) -> bool:
+    for order in orders or []:
+        if fill_group.order_id and str(order.get("order_id") or "") == fill_group.order_id:
+            return True
+    return False
+
+
+def price_is_close(left: float, right: float) -> bool:
+    anchor = max(abs(right), 1.0)
+    return abs(left - right) <= max(anchor * 0.005, 25.0)
+
+
+def _parse_dt(value) -> Optional[datetime]:
+    if value in (None, ""):
+        return None
+    if isinstance(value, datetime):
+        return _ensure_aware(value)
+    return _ensure_aware(datetime.fromisoformat(str(value).replace("Z", "+00:00")))
+
+
+def _ensure_aware(value: datetime) -> datetime:
+    if value.tzinfo is None:
+        return value.replace(tzinfo=timezone.utc)
+    return value.astimezone(timezone.utc)

src/services/runtime/trading_runtime_service.py

@@ -10,6 +10,7 @@ from src.config.settings import settings
 from src.services.broker import BROKER_STATE_OK, BrokerReadModelService
 from src.services.decision import DecisionPacketService
 from src.services.market_state import MarketStateService, get_market_truth_supervisor
+from src.services.reconciliation import BrokerExecutionReconciliationService
 from src.services.trade_plan_candidates import SetupWatcherService
 from src.services.trading_plans import TradingPlanService
 
@@ -54,12 +55,16 @@ class TradingRuntimeScheduler:
         trading_plan_service: Optional[TradingPlanService] = None,
         setup_watcher_service: Optional[SetupWatcherService] = None,
         broker_read_model_service: Optional[BrokerReadModelService] = None,
+        reconciliation_service: Optional[BrokerExecutionReconciliationService] = None,
         decision_packet_service: Optional[DecisionPacketService] = None,
     ) -> None:
         self.market_state_service = market_state_service or MarketStateService(market_truth_supervisor=get_market_truth_supervisor())
         self.trading_plan_service = trading_plan_service or TradingPlanService()
         self.setup_watcher_service = setup_watcher_service or SetupWatcherService()
         self.broker_read_model_service = broker_read_model_service or BrokerReadModelService()
+        self.reconciliation_service = reconciliation_service or BrokerExecutionReconciliationService(
+            broker_read_model_service=self.broker_read_model_service,
+        )
         self.decision_packet_service = decision_packet_service or DecisionPacketService()
         self.symbols = [item.strip() for item in settings.trading_runtime_symbols.split(",") if item.strip()]
         self.bias_tf = settings.trading_runtime_bias_timeframe
@@ -68,6 +73,7 @@ class TradingRuntimeScheduler:
         self.tasks = {
             "market_state": RuntimeTaskState("market_state", settings.trading_runtime_market_state_interval_seconds),
             "broker_sync": RuntimeTaskState("broker_sync", settings.trading_runtime_broker_sync_interval_seconds),
+            "reconciliation": RuntimeTaskState("reconciliation", settings.trading_runtime_reconciliation_interval_seconds),
             "watch": RuntimeTaskState("watch", settings.trading_runtime_watch_interval_seconds),
             "decision_packet": RuntimeTaskState("decision_packet", settings.trading_runtime_decision_packet_interval_seconds),
             "plan": RuntimeTaskState("plan", settings.trading_runtime_plan_interval_seconds),
@@ -97,6 +103,7 @@ class TradingRuntimeScheduler:
             self._run_plan_task(symbol=symbol, instrument_id=instrument_id, now=now, blockers=blockers)
             market_payload = self._run_market_state_task(symbol=symbol, instrument_id=instrument_id, now=now, blockers=blockers)
             self._run_broker_sync_task(symbol=symbol, instrument_id=instrument_id, now=now, blockers=blockers)
+            self._run_reconciliation_task(symbol=symbol, instrument_id=instrument_id, now=now, blockers=blockers)
             self._run_watch_task(symbol=symbol, instrument_id=instrument_id, now=now, market_payload=market_payload, blockers=blockers)
             self._run_decision_packet_task(symbol=symbol, instrument_id=instrument_id, now=now, blockers=blockers)
         self.current_blockers = _dedupe(blockers)
@@ -241,6 +248,32 @@ class TradingRuntimeScheduler:
             blockers.append(code)
             self._record_incident(task_name=task.name, symbol=symbol, code=code, error=str(exc), observed_ts=now)
 
+    def _run_reconciliation_task(self, *, symbol: str, instrument_id: int, now: datetime, blockers: list[str]) -> None:
+        task = self.tasks["reconciliation"]
+        if not _task_due(task, now):
+            return
+        _task_start(task, now)
+        try:
+            summary = self.reconciliation_service.reconcile_instrument(instrument_id=instrument_id, now=now)
+            task.details = {
+                "symbol": symbol,
+                "instrument_id": instrument_id,
+                "status": summary.get("status"),
+                "matched_execution_ids": list(summary.get("matched_execution_ids") or []),
+                "imported_execution_ids": list(summary.get("imported_execution_ids") or []),
+                "journal_updates": list(summary.get("journal_updates") or []),
+            }
+            if summary.get("status") != "ok":
+                code = f"reconciliation_degraded:{symbol}:{summary.get('error') or summary.get('status')}"
+                blockers.append(code)
+                self._record_incident(task_name=task.name, symbol=symbol, code=code, error=str(summary.get("error")), observed_ts=now)
+            _task_success(task, now)
+        except Exception as exc:
+            _task_failure(task, now, str(exc))
+            code = f"reconciliation_task_failed:{symbol}"
+            blockers.append(code)
+            self._record_incident(task_name=task.name, symbol=symbol, code=code, error=str(exc), observed_ts=now)
+
     def _run_decision_packet_task(self, *, symbol: str, instrument_id: int, now: datetime, blockers: list[str]) -> None:
         task = self.tasks["decision_packet"]
         if not _task_due(task, now):

src/web/app.js

@@ -546,6 +546,16 @@ function renderTickets(items) {
 
 function renderExecutions(items) {
   clearNode(positionManagementPanel);
+  positionManagementPanel.appendChild(
+    buildActionBar([
+      {
+        label: 'Reconcile Real Trades',
+        onClick: () => runAction('/trading/reconcile', {
+          instrument_id: Number(instrumentIdInput.value || 1),
+        }),
+      },
+    ])
+  );
   const filtered = filterExecutions(items);
   if (!filtered.length) {
     positionManagementPanel.appendChild(createElement('div', 'empty-state', 'No execution rows match the current filters.'));
@@ -574,6 +584,7 @@ function renderExecutions(items) {
           ['opened', formatTimestamp(execution.opened_ts)],
           ['closed', formatTimestamp(execution.closed_ts)],
           ['quality', formatNumber(execution.quality_score)],
+          ['real trade mapping', formatJsonInline(execution.reconciliation_state || {})],
           ['bracket state', formatJsonInline(execution.bracket_state || {})],
           ['reconciliation', formatJsonInline(execution.reconciliation_state || {})],
         ],
@@ -611,9 +622,12 @@ function renderPostTradeReview(summary, journalItems, weeklyReview) {
 
   const top = buildInfoGrid([
     ['entry count', String(summary?.entry_count ?? weeklyReview?.entry_count ?? 0)],
+    ['real trades', String(weeklyReview?.real_trade_count ?? 0)],
     ['framework execution rate', formatPercent(weeklyReview?.framework_execution_rate)],
     ['out-of-model rate', formatPercent(weeklyReview?.out_of_model_trade_rate)],
+    ['out-of-model real trades', formatPercent(weeklyReview?.out_of_model_real_trade_rate)],
     ['risk violation count', String(weeklyReview?.risk_violation_count ?? 0)],
+    ['reconciliation issues', String(weeklyReview?.reconciliation_issue_count ?? 0)],
     ['high quality session', formatJsonInline(weeklyReview?.high_quality_session)],
     ['low quality session', formatJsonInline(weeklyReview?.low_quality_session)],
     ['next actions', formatList(weeklyReview?.next_week_actions || [])],

tests/unit/test_api_server.py

@@ -68,6 +68,18 @@ class ApiServerRoutingTests(unittest.TestCase):
         self.assertEqual(status_alerts, 200)
         self.assertEqual(payload_alerts["items"], [{"id": 1, "alert_type": "candidate_actionable"}])
 
+    def test_routes_trading_reconcile(self) -> None:
+        status, payload = route_request(
+            "/trading/reconcile",
+            {},
+            method="POST",
+            body={"instrument_id": 1},
+            dependencies={"trading_reconciliation_run": lambda **kwargs: {"status": "ok", "imported_execution_ids": [7]}},
+        )
+
+        self.assertEqual(status, 200)
+        self.assertEqual(payload["imported_execution_ids"], [7])
+
     def test_routes_trading_broker_state(self) -> None:
         status, payload = route_request(
             "/trading/broker-state",

tests/unit/test_broker_execution_reconciliation_service.py

@@ -0,0 +1,222 @@
+from datetime import datetime, timezone
+import os
+import unittest
+
+os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://user:pass@localhost:5432/ai_ict_test")
+
+from src.services.reconciliation import BrokerExecutionReconciliationService
+from src.services.reconciliation.broker_execution_reconciliation_service import (
+    BROKER_STATE_OK,
+    BrokerFillGroup,
+    build_fill_groups,
+)
+
+
+class BrokerExecutionReconciliationServiceTests(unittest.TestCase):
+    def test_reconcile_matches_existing_execution_by_order_id(self) -> None:
+        repository = _ExecutionRepo(
+            rows=[
+                {
+                    "execution_id": 11,
+                    "signal_id": 21,
+                    "execution_ticket_id": 31,
+                    "venue_order_id": None,
+                    "execution_mode": "semi_auto",
+                    "execution_status": "pending",
+                    "side": "buy",
+                    "entry_price": 100.0,
+                    "size": 1.0,
+                    "opened_ts": datetime(2026, 5, 8, 0, 0, tzinfo=timezone.utc),
+                    "closed_ts": None,
+                    "broker_order_ids": ["ord-1"],
+                    "fill_state": "pending_submit",
+                    "position_state": "pending_entry",
+                    "close_reason": None,
+                    "reconciliation_state": {},
+                    "execution_meta": {},
+                    "instrument_id": 1,
+                    "model_code": "OSOK",
+                    "session_code": "LONDON",
+                    "signal_meta": {},
+                }
+            ],
+            executions={
+                11: {
+                    "id": 11,
+                    "signal_id": 21,
+                    "execution_ticket_id": 31,
+                    "status": "pending",
+                    "fill_state": "pending_submit",
+                    "position_state": "pending_entry",
+                    "close_reason": None,
+                    "reconciliation_state": {},
+                    "bracket_state": {},
+                    "broker_order_ids": ["ord-1"],
+                    "confirmed_by": "alice",
+                    "confirmed_at": datetime(2026, 5, 8, 0, 0, tzinfo=timezone.utc),
+                    "confirmation_required": True,
+                    "quality_score": 80.0,
+                    "ai_explanation": "ready",
+                    "meta": {},
+                    "closed_ts": None,
+                }
+            },
+        )
+        signal_repository = _SignalRepo()
+        journal_service = _JournalService()
+        broker_service = _BrokerService(
+            fills=[
+                {
+                    "fill_id": "fill-1",
+                    "order_id": "ord-1",
+                    "instrument_id": "BTC-USDT-SWAP",
+                    "side": "buy",
+                    "fill_size": 1.0,
+                    "fill_price": 100.0,
+                    "fill_ts": datetime(2026, 5, 8, 0, 1, tzinfo=timezone.utc),
+                }
+            ],
+            positions=[{"instrument_id": "BTC-USDT-SWAP", "side": "buy", "size": 1.0}],
+            orders=[],
+        )
+        service = BrokerExecutionReconciliationService(
+            execution_repository=repository,
+            signal_repository=signal_repository,
+            broker_read_model_service=broker_service,
+            journal_service=journal_service,
+        )
+
+        summary = service.reconcile_instrument(instrument_id=1, now=datetime(2026, 5, 8, 0, 2, tzinfo=timezone.utc))
+
+        self.assertEqual(summary["status"], "ok")
+        self.assertEqual(summary["matched_execution_ids"], [11])
+        self.assertEqual(summary["imported_execution_ids"], [])
+        self.assertEqual(journal_service.execution_ids, [11])
+        self.assertEqual(repository.updated[0]["reconciliation_state"]["status"], "matched_fill")
+        self.assertEqual(repository.updated[0]["status"], "open")
+
+    def test_reconcile_imports_unmatched_fill_group(self) -> None:
+        repository = _ExecutionRepo(rows=[], executions={})
+        signal_repository = _SignalRepo()
+        journal_service = _JournalService()
+        broker_service = _BrokerService(
+            fills=[
+                {
+                    "fill_id": "fill-2",
+                    "order_id": "ord-2",
+                    "instrument_id": "BTC-USDT-SWAP",
+                    "side": "sell",
+                    "fill_size": 0.5,
+                    "fill_price": 105.0,
+                    "fill_ts": datetime(2026, 5, 8, 2, 1, tzinfo=timezone.utc),
+                }
+            ],
+            positions=[],
+            orders=[],
+        )
+        service = BrokerExecutionReconciliationService(
+            execution_repository=repository,
+            signal_repository=signal_repository,
+            broker_read_model_service=broker_service,
+            journal_service=journal_service,
+        )
+
+        summary = service.reconcile_instrument(instrument_id=1, now=datetime(2026, 5, 8, 2, 2, tzinfo=timezone.utc))
+
+        self.assertEqual(summary["status"], "ok")
+        self.assertEqual(summary["matched_execution_ids"], [])
+        self.assertEqual(summary["imported_execution_ids"], [901])
+        self.assertEqual(signal_repository.created_signal_ids, [801])
+        self.assertEqual(journal_service.execution_ids, [901])
+
+    def test_build_fill_groups_merges_partial_fills_by_order(self) -> None:
+        groups = build_fill_groups(
+            [
+                {
+                    "fill_id": "f-1",
+                    "order_id": "ord-1",
+                    "instrument_id": "BTC-USDT-SWAP",
+                    "side": "buy",
+                    "fill_size": 1.0,
+                    "fill_price": 100.0,
+                    "fill_ts": datetime(2026, 5, 8, 1, 0, tzinfo=timezone.utc),
+                },
+                {
+                    "fill_id": "f-2",
+                    "order_id": "ord-1",
+                    "instrument_id": "BTC-USDT-SWAP",
+                    "side": "buy",
+                    "fill_size": 2.0,
+                    "fill_price": 101.0,
+                    "fill_ts": datetime(2026, 5, 8, 1, 1, tzinfo=timezone.utc),
+                },
+            ]
+        )
+
+        self.assertEqual(len(groups), 1)
+        self.assertEqual(groups[0].total_size, 3.0)
+        self.assertAlmostEqual(groups[0].avg_price, (1.0 * 100.0 + 2.0 * 101.0) / 3.0)
+
+
+class _ExecutionRepo:
+    def __init__(self, *, rows, executions):
+        self.rows = rows
+        self.executions = executions
+        self.updated = []
+        self.inserted = []
+
+    def fetch_recent_execution_reconciliation_rows(self, instrument_id: int, limit: int = 200):
+        return list(self.rows)
+
+    def fetch_execution_by_id(self, execution_id: int):
+        return dict(self.executions.get(execution_id) or {})
+
+    def update_trade_execution_lifecycle(self, **kwargs):
+        self.updated.append(kwargs)
+
+    def insert_trade_execution(self, **kwargs):
+        self.inserted.append(kwargs)
+        return 901
+
+
+class _SignalRepo:
+    def __init__(self):
+        self.created_signal_ids = []
+
+    def fetch_session_at(self, instrument_id: int, ts):
+        return {"session_code": "LONDON"}
+
+    def insert_trade_signal_returning_id(self, **kwargs):
+        self.created_signal_ids.append(801)
+        return 801
+
+
+class _JournalService:
+    def __init__(self):
+        self.execution_ids = []
+
+    def upsert_entry_for_execution(self, execution_id: int):
+        self.execution_ids.append(execution_id)
+        return len(self.execution_ids)
+
+
+class _BrokerSnapshot:
+    status = BROKER_STATE_OK
+    error = None
+
+    def __init__(self, *, fills, positions, orders):
+        self.fills = fills
+        self.positions = positions
+        self.orders = orders
+
+
+class _BrokerService:
+    def __init__(self, *, fills, positions, orders):
+        self.snapshot = _BrokerSnapshot(fills=fills, positions=positions, orders=orders)
+
+    def fetch_snapshot(self, **kwargs):
+        return self.snapshot
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/unit/test_runtime_script_json.py

@@ -19,6 +19,7 @@ from scripts import (
     run_displacement,
     run_fvg,
     run_liquidity,
+    run_broker_reconciliation,
     run_mss,
     run_ranges,
     run_decision_packet,
@@ -135,6 +136,18 @@ class RuntimeScriptJsonTests(unittest.TestCase):
         self.assertEqual(exit_code, 0)
         self.assertEqual(payload["instrument_id"], 2)
 
+    def test_run_broker_reconciliation_outputs_json(self) -> None:
+        service = type("Service", (), {"reconcile_instrument": lambda self, instrument_id: {"status": "ok", "instrument_id": instrument_id, "matched_execution_ids": [1]}})()
+        output = io.StringIO()
+        with patch("scripts.run_broker_reconciliation.BrokerExecutionReconciliationService", return_value=service), contextlib.redirect_stdout(output):
+            exit_code = run_broker_reconciliation.main(
+                env={"BROKER_RECONCILIATION_OUTPUT_JSON": "1", "BROKER_RECONCILIATION_INSTRUMENT_ID": "2"}
+            )
+
+        payload = json.loads(output.getvalue())
+        self.assertEqual(exit_code, 0)
+        self.assertEqual(payload["instrument_id"], 2)
+
 
 if __name__ == "__main__":
     unittest.main()

tests/unit/test_trading_journal_service.py

@@ -25,6 +25,8 @@ class TradingJournalServiceTests(unittest.TestCase):
         self.assertTrue(entry.plan_violation)
         self.assertTrue(entry.risk_violation)
         self.assertFalse(entry.framework_execution)
+        self.assertTrue(entry.real_trade_detected)
+        self.assertFalse(entry.out_of_model_trade)
         self.assertIn("plan_violation", entry.review_tags)
         self.assertIn("tighten_plan_and_candidate_validation", entry.next_actions)
 
@@ -45,6 +47,9 @@ class TradingJournalServiceTests(unittest.TestCase):
                 "framework_execution": True,
                 "model_compliance": True,
                 "risk_violation": False,
+                "real_trade_detected": True,
+                "out_of_model_trade": False,
+                "reconciliation_status": "matched_fill",
                 "next_actions": ["keep_same_model"],
                 "summary": {"backtest_result": {"meta": {"r_multiple": 1.2}}},
             },
@@ -53,6 +58,9 @@ class TradingJournalServiceTests(unittest.TestCase):
                 "framework_execution": False,
                 "model_compliance": False,
                 "risk_violation": True,
+                "real_trade_detected": True,
+                "out_of_model_trade": True,
+                "reconciliation_status": "imported_external_fill",
                 "next_actions": ["review_risk_process"],
                 "summary": {"backtest_result": {"meta": {"r_multiple": -0.8}}},
             },
@@ -63,6 +71,7 @@ class TradingJournalServiceTests(unittest.TestCase):
         )
 
         self.assertEqual(summary["entry_count"], 2)
+        self.assertEqual(summary["real_trade_count"], 2)
         self.assertEqual(summary["risk_violation_count"], 1)
         self.assertEqual(summary["high_quality_session"]["session_code"], "LONDON")
         self.assertEqual(summary["low_quality_session"]["session_code"], "ASIA")
@@ -131,6 +140,11 @@ def _bundle() -> dict:
         "execution_status": "closed",
         "opened_ts": datetime(2026, 5, 6, 0, 0, tzinfo=timezone.utc),
         "closed_ts": datetime(2026, 5, 6, 1, 0, tzinfo=timezone.utc),
+        "broker_order_ids": ["ord-1"],
+        "fill_state": "filled",
+        "position_state": "closed",
+        "close_reason": "tp2_closed",
+        "reconciliation_state": {"status": "matched_fill", "broker_fill_ids": ["fill-1"], "fill_count": 1},
         "execution_meta": {"risk_checks": ["missing_structural_targets", "news_window_active"]},
         "instrument_id": 1,
         "model_code": "SWEEP_MSS_FVG",

tests/unit/test_trading_runtime_service.py

@@ -95,6 +95,16 @@ class FakeDecisionPacketService:
         }
 
 
+class FakeReconciliationService:
+    def reconcile_instrument(self, **kwargs):
+        return {
+            "status": "ok",
+            "matched_execution_ids": [91],
+            "imported_execution_ids": [92],
+            "journal_updates": [11, 12],
+        }
+
+
 class TradingRuntimeSchedulerTests(unittest.TestCase):
     def test_run_cycle_records_successful_task_state(self) -> None:
         scheduler = TradingRuntimeScheduler(
@@ -102,6 +112,7 @@ class TradingRuntimeSchedulerTests(unittest.TestCase):
             trading_plan_service=FakeTradingPlanService(),
             setup_watcher_service=FakeSetupWatcherService(),
             broker_read_model_service=FakeBrokerReadModelService(),
+            reconciliation_service=FakeReconciliationService(),
             decision_packet_service=FakeDecisionPacketService(),
         )
         with patch("src.services.runtime.trading_runtime_service.settings.trading_runtime_symbols", "BTC-USDT"):
@@ -111,6 +122,7 @@ class TradingRuntimeSchedulerTests(unittest.TestCase):
         self.assertEqual(payload["current_blockers"], [])
         self.assertEqual(payload["tasks"]["plan"]["last_status"], "ok")
         self.assertEqual(payload["tasks"]["broker_sync"]["details"]["status"], "broker_state_ok")
+        self.assertEqual(payload["tasks"]["reconciliation"]["details"]["matched_execution_ids"], [91])
         self.assertEqual(payload["tasks"]["decision_packet"]["details"]["packet_id"], 501)
         self.assertEqual(payload["latest_broker_state_by_symbol"]["BTC-USDT"]["account"]["equity"], 1000.0)
         self.assertTrue(payload["tasks"]["watch"]["next_run_ts"])
@@ -121,6 +133,7 @@ class TradingRuntimeSchedulerTests(unittest.TestCase):
             trading_plan_service=FakeTradingPlanService(),
             setup_watcher_service=FakeSetupWatcherService(),
             broker_read_model_service=FakeBrokerReadModelService(),
+            reconciliation_service=FakeReconciliationService(),
             decision_packet_service=FakeDecisionPacketService(),
         )
         with patch("src.services.runtime.trading_runtime_service.settings.trading_runtime_symbols", "BTC-USDT"):
@@ -137,6 +150,7 @@ class TradingRuntimeSchedulerTests(unittest.TestCase):
             trading_plan_service=FakeTradingPlanService(),
             setup_watcher_service=FakeSetupWatcherService(),
             broker_read_model_service=FakeBrokerReadModelService(),
+            reconciliation_service=FakeReconciliationService(),
             decision_packet_service=FakeDecisionPacketService(),
         )
         supervisor = TradingRuntimeSupervisor(scheduler=scheduler)
@@ -167,6 +181,18 @@ class TradingRuntimeApiTests(unittest.TestCase):
         self.assertEqual(run_status, 200)
         self.assertIn("tasks", run_payload)
 
+    def test_routes_reconciliation_run(self) -> None:
+        status, payload = route_request(
+            "/trading/reconcile",
+            {},
+            method="POST",
+            body={"instrument_id": 1},
+            dependencies={"trading_reconciliation_run": lambda **kwargs: {"status": "ok", "matched_execution_ids": [1]}},
+        )
+
+        self.assertEqual(status, 200)
+        self.assertEqual(payload["matched_execution_ids"], [1])
+
 
 class TradingRuntimeScriptTests(unittest.TestCase):
     def test_script_outputs_json(self) -> None: