Signal processing pipeline: bar-by-bar execution

Data source (CSV / Bitget / OANDA / sample)
    ↓  data_connectors.py — normalizes to [{time, open, high, low, close, volume}]
    ↓  SMKPipeline.load_bars()
    ↓  SMKPipeline.step()  — called once per bar
    ├── Layer 1 (Structural):  core/detectors/ — DealingRange, BiasDetector, IPDAPhase,
    │                          EquilibriumCross, SessionDetector, SwingDetector
    ├── Layer 2 (Memory):      liquidity/ — FVGDetector, OrderBlockDetector, VolumeProfile
    ├── Layer 3 (λ Sensors):   lambda_sensors/ — λ₁Volatility, λ₂Killzone, λ₃Harmonic,
    │                          λ₄Manipulation, λ₅Displacement, λ₆MacroBias
    ├── Layer 4 / Ring 0:      detectors/ + risk/ — KLDivergence, TopologicalFracture,
    │                          MandraKernel, LambdaFusionEngine
    ├── Plugin layer:          backend/plugins/ — 6 forensic plugins appended to sensors[]
    └── AEGIS bridge:          backend/aegis_bridge.py — StopLossManager + SchurRouter
    ↓  step() returns a single result dict
WebSocket /ws/stream  →  frontend/index.html (TradingView Lightweight Charts)

import asyncio
from backend.smk_pipeline import SMKPipeline
from backend.data_connectors import load_csv

pipeline = SMKPipeline()
bars = load_csv("EURUSD_H1.csv")
pipeline.load_bars(bars)

async def run():
    while True:
        result = await pipeline.step()
        if result is None:
            break  # All bars consumed

        # Check veto before acting
        if result["veto"]["trade_allowed"]:
            print(f"Bar {result['bar_index']}: PROCEED — p_fused={result['fusion']['p_fused']:.3f}")
        else:
            reasons = result["veto"]["reasons"]
            print(f"Bar {result['bar_index']}: HALT — {reasons}")

asyncio.run(run())

result["veto"] == {
    "decision":      "Proceed",   # "Proceed" | "Halt" | "Reset"
    "trade_allowed": True,         # False if any Ring 0 condition fires
    "reasons":       []            # List of veto reason strings
}

def _load_modules(self):
    def try_load(key, factory):
        try:
            self.modules[key] = factory()
        except Exception as e:
            self._import_errors.append(f"{key}: {e}")
            self.modules[key] = None

    try_load("bias",    lambda: _imp("core.bias_detector", "BiasDetector")())
    try_load("fusion",  lambda: _imp("core.kernel.lambda_fusion_engine", "LambdaFusionEngine")())
    try_load("mandra",  lambda: _imp("risk.mandra_kernels", "MandraGate")())
    # ... 20+ more modules

def _sanitize(obj):
    """Recursively convert numpy types to plain Python before JSON serialization."""
    if isinstance(obj, dict):
        return {k: _sanitize(v) for k, v in obj.items()}
    if isinstance(obj, list):
        return [_sanitize(v) for v in obj]
    if isinstance(obj, (np.integer,)):
        return int(obj)
    if isinstance(obj, (np.floating,)):
        return float(obj)
    if isinstance(obj, np.ndarray):
        return obj.tolist()
    return obj