Add RL trading strategy, regime-based MDF, dynamic strategy switching

- Integrate Adilbai/stock-trading-rl-agent PPO model as alternative CH
trading strategy (new ch_rl_trader.py)
- CH_AI_STRATEGY: llm/rl/hybrid (default hybrid, dynamically switchable
via POST /ch/api/strategy)
- MDF now uses regime-driven price dynamics (trending/mean-reverting/
volatile/calm) producing meaningful SMA, EMA, MACD, RSI, Bollinger
Band signals embedded in snapshots
- Rename source CLEARINGHOUSE → CLRH in order messages
- Add stable-baselines3, pandas, scikit-learn, huggingface_hub to Docker

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Dockerfile

@@ -33,7 +33,12 @@ RUN pip install --no-cache-dir \
       kafka-python==2.0.2 \
       Flask==2.2.5 \
       requests==2.31.0 \
-      quickfix
+      quickfix \
+      numpy \
+      pandas \
+      scikit-learn \
+      "stable-baselines3[extra]" \
+      huggingface_hub
 
 # ── Application code (flat layout matching /app container paths) ──────────────
 WORKDIR /app

clearing_house/Dockerfile

@@ -1,6 +1,7 @@
 FROM python:3.11-slim
 
-RUN pip install --no-cache-dir flask kafka-python requests
+RUN pip install --no-cache-dir flask kafka-python requests \
+      numpy pandas scikit-learn "stable-baselines3[extra]" huggingface_hub
 
 WORKDIR /app
 

clearing_house/app.py

@@ -258,7 +258,7 @@ def submit_order():
         "ord_type":     "LIMIT",
         "time_in_force": "DAY",
         "timestamp":    time.time(),
-        "source":       "CLEARINGHOUSE",
+        "source":       "CLRH",
     }
     get_producer().send(Config.ORDERS_TOPIC, msg)
     return jsonify({"status": "ok", "cl_ord_id": cl_ord_id})
@@ -401,6 +401,24 @@ def api_market():
     return jsonify(_get_bbos())
 
 
+@app.route("/ch/api/config")
+def api_config():
+    return jsonify({
+        "strategy": ai_trader.get_strategy(),
+        "obligation": db.CH_DAILY_OBLIGATION,
+        "ai_interval": int(os.getenv("CH_AI_INTERVAL", "45")),
+    })
+
+
+@app.route("/ch/api/strategy", methods=["POST"])
+def api_set_strategy():
+    data = request.get_json(force=True)
+    strategy = data.get("strategy", "")
+    result = ai_trader.set_strategy(strategy)
+    _broadcast("config", {"strategy": result})
+    return jsonify({"status": "ok", "strategy": result})
+
+
 # ── SSE ────────────────────────────────────────────────────────────────────────
 
 @app.route("/ch/stream")

clearing_house/ch_ai_trader.py

@@ -1,10 +1,13 @@
 """AI-driven simulation of CH members not currently controlled by a real user.
 
-Two background threads:
+Three background threads:
   1. _trade_consumer_thread  – Kafka consumer on 'trades' topic; attributes
      trades back to CH members whose cl_ord_id starts with USRxx-.
   2. _simulation_thread      – every CH_AI_INTERVAL seconds, picks an order
-     for each unoccupied member using the LLM (Groq → HF → Ollama fallback).
+     for each unoccupied member using the configured strategy.
+  3. _control_listener_thread – listens for session start/stop/suspend/resume.
+
+Strategy is selected via CH_AI_STRATEGY env var: "llm", "rl", or "hybrid".
 
 Call start() once from app.py after init_db().
 Call set_human_active(member_id) / set_human_inactive(member_id) on login/logout.
@@ -28,9 +31,18 @@ from shared.kafka_utils import create_consumer, create_producer
 
 import ch_database as db
 
+# RL trader (optional – graceful fallback if deps not installed)
+try:
+    import ch_rl_trader as rl_trader
+    _rl_available = True
+except ImportError:
+    _rl_available = False
+    print("[CH-AI] RL dependencies not installed — RL strategy unavailable")
+
 # ── Config ─────────────────────────────────────────────────────────────────────
 CH_AI_INTERVAL = int(os.getenv("CH_AI_INTERVAL", "45"))   # seconds between AI cycles
-CH_SOURCE      = "CLEARINGHOUSE"
+CH_AI_STRATEGY = os.getenv("CH_AI_STRATEGY", "hybrid")     # "llm", "rl", or "hybrid"
+CH_SOURCE      = "CLRH"
 
 OLLAMA_HOST  = os.getenv("OLLAMA_HOST", "")
 OLLAMA_MODEL = os.getenv("OLLAMA_MODEL", "llama3.1:8b")
@@ -71,6 +83,24 @@ def is_human_active(member_id: str) -> bool:
         return member_id in _active_humans
 
 
+def get_strategy() -> str:
+    return CH_AI_STRATEGY
+
+
+def set_strategy(strategy: str) -> str:
+    """Dynamically switch AI strategy. Returns the active strategy."""
+    global CH_AI_STRATEGY
+    strategy = strategy.lower().strip()
+    if strategy not in ("llm", "rl", "hybrid"):
+        return CH_AI_STRATEGY
+    if strategy in ("rl", "hybrid") and not _rl_available:
+        print(f"[CH-AI] Cannot switch to {strategy}: RL deps not installed")
+        return CH_AI_STRATEGY
+    CH_AI_STRATEGY = strategy
+    print(f"[CH-AI] Strategy switched to: {strategy}")
+    return CH_AI_STRATEGY
+
+
 def start() -> None:
     """Start background threads. Call once after app startup."""
     global _running
@@ -78,7 +108,11 @@ def start() -> None:
     threading.Thread(target=_trade_consumer_thread, daemon=True, name="ch-trade-consumer").start()
     threading.Thread(target=_simulation_thread,     daemon=True, name="ch-ai-sim").start()
     threading.Thread(target=_control_listener_thread, daemon=True, name="ch-control").start()
-    print("[CH-AI] Background threads started")
+    strategy = CH_AI_STRATEGY
+    if strategy in ("rl", "hybrid") and not _rl_available:
+        strategy = "llm"
+        print("[CH-AI] WARNING: RL requested but deps missing, falling back to LLM")
+    print(f"[CH-AI] Background threads started (strategy={strategy})")
 
 
 # ── Kafka helpers ──────────────────────────────────────────────────────────────
@@ -127,6 +161,13 @@ def _trade_consumer_thread():
             if not symbol or price <= 0 or qty <= 0:
                 continue
 
+            # Feed every trade into RL price history (regardless of strategy)
+            if _rl_available:
+                try:
+                    rl_trader.feed_trade(symbol, price, qty)
+                except Exception:
+                    pass
+
             # Detect CH member orders by cl_ord_id prefix pattern USRxx-
             for order_id, side in [(buy_id, "BUY"), (sell_id, "SELL")]:
                 m = re.match(r"^(USR\d{2})-", order_id)
@@ -207,12 +248,44 @@ def _run_simulation_cycle():
         if obligation_remaining == 0 and random.random() > 0.3:
             continue  # occasionally trade even after obligation met
 
-        order = _decide_order_llm(mid, capital, holdings, dt, bbos, obligation_remaining)
+        order = _decide_order(mid, capital, holdings, dt, bbos, obligation_remaining)
         if order:
             _submit_order(mid, order)
             time.sleep(0.5)  # stagger submissions
 
 
+def _decide_order(member_id, capital, holdings, daily_trades, bbos, obligation_remaining):
+    """Dispatch to the configured strategy."""
+    strategy = CH_AI_STRATEGY
+
+    # Hybrid: split members between RL and LLM
+    if strategy == "hybrid" and _rl_available:
+        member_num = int(member_id[-2:])
+        strategy = "rl" if member_num <= 5 else "llm"
+
+    if strategy == "rl" and _rl_available:
+        try:
+            order = rl_trader.decide_order_rl(
+                member_id, capital, holdings, bbos, obligation_remaining,
+            )
+            if order and _validate_order(order, capital, holdings, bbos):
+                try:
+                    db.record_ai_decision(member_id, f"RL: {order}", order, source="rl")
+                except Exception:
+                    pass
+                return order
+        except Exception as e:
+            print(f"[CH-AI] RL strategy error for {member_id}: {e}")
+        # Fall through to LLM on RL failure
+        return _decide_order_llm(
+            member_id, capital, holdings, daily_trades, bbos, obligation_remaining,
+        )
+
+    return _decide_order_llm(
+        member_id, capital, holdings, daily_trades, bbos, obligation_remaining,
+    )
+
+
 def _load_reference_prices() -> dict:
     """Load reference prices from securities.txt as fallback when books are empty."""
     ref = {}

clearing_house/ch_rl_trader.py

@@ -0,0 +1,410 @@
+"""RL-based trading strategy using Adilbai/stock-trading-rl-agent (PPO).
+
+Provides decide_order_rl() with the same return type as _decide_order_llm()
+so it can be used as a drop-in alternative in ch_ai_trader.py.
+"""
+
+import os
+import pickle
+import random
+import threading
+import time
+from collections import deque
+from typing import Optional
+
+import numpy as np
+
+# ── Config ────────────────────────────────────────────────────────────────────
+RL_MODEL_REPO = os.getenv("CH_RL_MODEL_REPO", "Adilbai/stock-trading-rl-agent")
+RL_MODEL_CACHE = os.getenv("CH_RL_MODEL_CACHE", "/app/data/rl_model")
+RL_BAR_INTERVAL = int(os.getenv("CH_RL_BAR_INTERVAL", "60"))  # seconds per bar
+RL_MIN_BARS = int(os.getenv("CH_RL_MIN_BARS", "30"))          # min bars before RL kicks in
+RL_LOOKBACK = 60
+
+# ── Shared state ──────────────────────────────────────────────────────────────
+_model = None
+_scaler = None
+_model_lock = threading.Lock()
+_load_attempted = False
+
+# Per-symbol rolling price bars: {symbol: deque of {open, high, low, close, volume}}
+_price_bars: dict[str, deque] = {}
+_bars_lock = threading.Lock()
+# Accumulator for current bar being built
+_current_bar: dict[str, dict] = {}
+
+
+# ── Model loading ─────────────────────────────────────────────────────────────
+
+def _load_model():
+    """Download and load the PPO model + scaler from HuggingFace Hub."""
+    global _model, _scaler, _load_attempted
+    with _model_lock:
+        if _load_attempted:
+            return _model is not None
+        _load_attempted = True
+
+    try:
+        from huggingface_hub import hf_hub_download
+        from stable_baselines3 import PPO
+
+        os.makedirs(RL_MODEL_CACHE, exist_ok=True)
+        print(f"[CH-RL] Downloading model from {RL_MODEL_REPO}...")
+
+        model_path = hf_hub_download(
+            repo_id=RL_MODEL_REPO, filename="final_model.zip",
+            cache_dir=RL_MODEL_CACHE,
+        )
+        scaler_path = hf_hub_download(
+            repo_id=RL_MODEL_REPO, filename="scaler.pkl",
+            cache_dir=RL_MODEL_CACHE,
+        )
+
+        with _model_lock:
+            _model = PPO.load(model_path)
+            with open(scaler_path, "rb") as f:
+                _scaler = pickle.load(f)
+
+        print("[CH-RL] Model loaded successfully")
+        return True
+    except Exception as e:
+        print(f"[CH-RL] Failed to load model: {e}")
+        return False
+
+
+def is_available() -> bool:
+    """Check if RL model is loaded and ready."""
+    return _model is not None
+
+
+# ── Price history ─────────────────────────────────────────────────────────────
+
+def feed_trade(symbol: str, price: float, quantity: int):
+    """Feed a trade into the price bar accumulator. Called from trade consumer."""
+    with _bars_lock:
+        if symbol not in _current_bar:
+            _current_bar[symbol] = {
+                "open": price, "high": price, "low": price,
+                "close": price, "volume": quantity,
+                "start_time": time.time(),
+            }
+        else:
+            bar = _current_bar[symbol]
+            bar["high"] = max(bar["high"], price)
+            bar["low"] = min(bar["low"], price)
+            bar["close"] = price
+            bar["volume"] += quantity
+
+        # Finalize bar if interval elapsed
+        bar = _current_bar[symbol]
+        if time.time() - bar["start_time"] >= RL_BAR_INTERVAL:
+            _finalize_bar(symbol)
+
+
+def seed_price(symbol: str, ref_price: float):
+    """Seed initial bars from reference price when no trade history exists."""
+    with _bars_lock:
+        if symbol in _price_bars and len(_price_bars[symbol]) > 0:
+            return
+        if symbol not in _price_bars:
+            _price_bars[symbol] = deque(maxlen=120)
+        # Create flat bars with small noise for indicator computation
+        for i in range(RL_LOOKBACK):
+            noise = random.uniform(-0.02, 0.02) * ref_price
+            p = ref_price + noise
+            _price_bars[symbol].append({
+                "open": round(p, 2),
+                "high": round(p + abs(noise) * 0.5, 2),
+                "low": round(p - abs(noise) * 0.5, 2),
+                "close": round(p, 2),
+                "volume": random.randint(100, 500),
+            })
+
+
+def _finalize_bar(symbol: str):
+    """Move current accumulator into the history deque. Must hold _bars_lock."""
+    bar = _current_bar.pop(symbol, None)
+    if not bar:
+        return
+    if symbol not in _price_bars:
+        _price_bars[symbol] = deque(maxlen=120)
+    _price_bars[symbol].append({
+        "open": bar["open"], "high": bar["high"],
+        "low": bar["low"], "close": bar["close"],
+        "volume": bar["volume"],
+    })
+
+
+def get_bar_count(symbol: str) -> int:
+    with _bars_lock:
+        return len(_price_bars.get(symbol, []))
+
+
+# ── Technical indicators ──────────────────────────────────────────────────────
+
+def _compute_indicators(bars: list[dict]) -> np.ndarray:
+    """Compute technical indicators from OHLCV bars. Returns (n_bars, n_features) array."""
+    n = len(bars)
+    close = np.array([b["close"] for b in bars], dtype=np.float64)
+    high = np.array([b["high"] for b in bars], dtype=np.float64)
+    low = np.array([b["low"] for b in bars], dtype=np.float64)
+    opn = np.array([b["open"] for b in bars], dtype=np.float64)
+    volume = np.array([b["volume"] for b in bars], dtype=np.float64)
+
+    def sma(arr, w):
+        out = np.full(n, np.nan)
+        if n >= w:
+            cs = np.cumsum(arr)
+            out[w - 1:] = (cs[w - 1:] - np.concatenate([[0], cs[:-w]])) / w
+        # Fill NaN with first valid
+        for i in range(n):
+            if np.isnan(out[i]):
+                out[i] = arr[i]
+            else:
+                break
+        out[:] = np.where(np.isnan(out), out[np.argmax(~np.isnan(out))], out)
+        return out
+
+    def ema(arr, span):
+        out = np.empty(n)
+        alpha = 2.0 / (span + 1)
+        out[0] = arr[0]
+        for i in range(1, n):
+            out[i] = alpha * arr[i] + (1 - alpha) * out[i - 1]
+        return out
+
+    sma5 = sma(close, 5)
+    sma10 = sma(close, 10)
+    sma20 = sma(close, 20)
+    sma50 = sma(close, 50)
+    ema12 = ema(close, 12)
+    ema26 = ema(close, 26)
+
+    macd = ema12 - ema26
+    signal = ema(macd, 9)
+    histogram = macd - signal
+
+    # RSI
+    delta = np.diff(close, prepend=close[0])
+    gain = np.where(delta > 0, delta, 0.0)
+    loss = np.where(delta < 0, -delta, 0.0)
+    avg_gain = ema(gain, 14)
+    avg_loss = ema(loss, 14)
+    rs = np.where(avg_loss > 0, avg_gain / avg_loss, 100.0)
+    rsi = 100.0 - 100.0 / (1.0 + rs)
+
+    # Bollinger Bands
+    bb_mid = sma20
+    bb_std = np.full(n, 0.01)
+    for i in range(n):
+        start = max(0, i - 19)
+        bb_std[i] = max(np.std(close[start:i + 1]), 0.01)
+    bb_upper = bb_mid + 2 * bb_std
+    bb_lower = bb_mid - 2 * bb_std
+    bb_width = (bb_upper - bb_lower) / np.where(bb_mid > 0, bb_mid, 1.0)
+    bb_pos = (close - bb_lower) / np.where(bb_upper - bb_lower > 0, bb_upper - bb_lower, 1.0)
+
+    # Volatility
+    vol20 = np.full(n, 0.01)
+    for i in range(n):
+        start = max(0, i - 19)
+        vol20[i] = max(np.std(close[start:i + 1]), 0.01)
+
+    # Price changes
+    price_change = np.diff(close, prepend=close[0]) / np.where(close > 0, close, 1.0)
+    price_change_5 = np.zeros(n)
+    for i in range(5, n):
+        price_change_5[i] = (close[i] - close[i - 5]) / close[i - 5] if close[i - 5] > 0 else 0
+
+    hl_ratio = high / np.where(low > 0, low, 1.0)
+    oc_ratio = close / np.where(opn > 0, opn, 1.0)
+
+    vol_sma = sma(volume, 20)
+    vol_ratio = volume / np.where(vol_sma > 0, vol_sma, 1.0)
+
+    # Base features (20 columns)
+    features = np.column_stack([
+        close, volume, sma5, sma10, sma20, sma50,
+        ema12, ema26, rsi, macd, signal, histogram,
+        bb_upper, bb_lower, bb_width, bb_pos,
+        vol20, price_change, hl_ratio, vol_ratio,
+    ])
+
+    # Lagged features: close, volume, price_change, rsi, macd, vol20 at lags 1,2,3,5,10
+    lag_sources = np.column_stack([close, volume, price_change, rsi, macd, vol20])
+    lag_cols = []
+    for lag in [1, 2, 3, 5, 10]:
+        shifted = np.roll(lag_sources, lag, axis=0)
+        shifted[:lag] = lag_sources[0]  # fill with first value
+        lag_cols.append(shifted)
+    lags = np.hstack(lag_cols)  # (n, 30)
+
+    return np.hstack([features, lags])  # (n, 50)
+
+
+# ── Observation builder ───────────────────────────────────────────────────────
+
+def _build_observation(
+    symbol: str,
+    capital: float,
+    holdings: list,
+    bbos: dict,
+) -> Optional[np.ndarray]:
+    """Build the 3008-dim observation vector for one symbol."""
+    with _bars_lock:
+        bars = list(_price_bars.get(symbol, []))
+
+    if len(bars) < RL_MIN_BARS:
+        return None
+
+    # Pad to RL_LOOKBACK if needed
+    while len(bars) < RL_LOOKBACK:
+        bars.insert(0, bars[0])
+
+    bars = bars[-RL_LOOKBACK:]
+
+    indicators = _compute_indicators(bars)  # (60, 50)
+
+    # Scale using the loaded scaler
+    if _scaler is not None:
+        try:
+            indicators = _scaler.transform(indicators)
+        except Exception:
+            # Shape mismatch — normalize manually
+            mean = indicators.mean(axis=0)
+            std = indicators.std(axis=0)
+            std[std == 0] = 1.0
+            indicators = (indicators - mean) / std
+
+    market_state = indicators.flatten()  # 3000
+
+    # Portfolio state (8 features)
+    held_qty = 0
+    held_value = 0.0
+    for h in holdings:
+        if h["symbol"] == symbol:
+            held_qty = h["quantity"]
+            bbo = bbos.get(symbol, {})
+            price = bbo.get("best_bid") or h["avg_cost"]
+            held_value = held_qty * price
+            break
+
+    net_worth = capital + sum(
+        h["quantity"] * (bbos.get(h["symbol"], {}).get("best_bid") or h["avg_cost"])
+        for h in holdings
+    )
+    returns = (net_worth - 100_000) / 100_000  # relative to starting capital
+
+    portfolio_state = np.array([
+        capital, held_qty, net_worth, returns,
+        held_value, float(held_qty > 0),
+        capital / max(net_worth, 1.0),
+        held_value / max(net_worth, 1.0),
+    ], dtype=np.float64)
+
+    obs = np.concatenate([market_state, portfolio_state])
+    return obs.astype(np.float32)
+
+
+# ── Main decision function ────────────────────────────────────────────────────
+
+def decide_order_rl(
+    member_id: str,
+    capital: float,
+    holdings: list,
+    bbos: dict,
+    obligation_remaining: int,
+) -> Optional[dict]:
+    """Use the RL model to decide a trade. Returns order dict or None."""
+    if not _model:
+        if not _load_model():
+            return None
+
+    # Seed price history from BBOs for symbols we haven't seen
+    for sym, bbo in bbos.items():
+        mid = None
+        if bbo.get("best_bid") and bbo.get("best_ask"):
+            mid = (bbo["best_bid"] + bbo["best_ask"]) / 2
+        elif bbo.get("best_bid"):
+            mid = bbo["best_bid"]
+        elif bbo.get("best_ask"):
+            mid = bbo["best_ask"]
+        if mid:
+            seed_price(sym, mid)
+
+    # Run prediction for each symbol, pick the best actionable one
+    candidates = []
+
+    for sym, bbo in bbos.items():
+        obs = _build_observation(sym, capital, holdings, bbos)
+        if obs is None:
+            continue
+
+        try:
+            action, _ = _model.predict(obs, deterministic=False)
+            action_type = int(round(float(action[0])))
+            position_size = float(np.clip(action[1], 0.05, 0.5))
+
+            action_type = max(0, min(2, action_type))
+
+            if action_type == 0:
+                continue  # Hold
+
+            if action_type == 1:  # Buy
+                ask = bbo.get("best_ask")
+                if not ask or ask <= 0:
+                    continue
+                affordable = int(capital // ask)
+                qty = max(10, int(affordable * position_size))
+                qty = min(qty, 200)
+                if qty * ask > capital:
+                    qty = int(capital // ask)
+                if qty < 10:
+                    continue
+                candidates.append({
+                    "symbol": sym, "side": "BUY",
+                    "quantity": qty, "price": round(ask, 2),
+                    "score": position_size,
+                })
+
+            elif action_type == 2:  # Sell
+                held = next((h["quantity"] for h in holdings if h["symbol"] == sym), 0)
+                if held <= 0:
+                    continue
+                bid = bbo.get("best_bid")
+                if not bid or bid <= 0:
+                    continue
+                qty = max(10, int(held * position_size))
+                qty = min(qty, held, 200)
+                if qty < 10:
+                    continue
+                candidates.append({
+                    "symbol": sym, "side": "SELL",
+                    "quantity": qty, "price": round(bid, 2),
+                    "score": position_size,
+                })
+        except Exception as e:
+            print(f"[CH-RL] Prediction error for {sym}: {e}")
+            continue
+
+    if not candidates:
+        # If obligation remains, force a buy on a random affordable symbol
+        if obligation_remaining > 0:
+            affordable = [
+                (sym, bbo) for sym, bbo in bbos.items()
+                if bbo.get("best_ask") and 10 * bbo["best_ask"] <= capital
+            ]
+            if affordable:
+                sym, bbo = random.choice(affordable)
+                return {
+                    "symbol": sym, "side": "BUY",
+                    "quantity": random.randint(10, 50),
+                    "price": round(bbo["best_ask"], 2),
+                }
+        return None
+
+    # Pick highest-confidence candidate
+    candidates.sort(key=lambda c: c["score"], reverse=True)
+    best = candidates[0]
+    best.pop("score")
+    return best

docker-compose.yml

@@ -208,6 +208,7 @@ services:
       - GROQ_API_KEY=${GROQ_API_KEY:-}
       - GROQ_MODEL=${GROQ_MODEL:-llama-3.1-8b-instant}
       - OLLAMA_HOST=${OLLAMA_HOST:-}
+      - CH_AI_STRATEGY=${CH_AI_STRATEGY:-hybrid}
     extra_hosts:
       - "host.docker.internal:host-gateway"
 

md_feeder/mdf_simulator.py

@@ -1,8 +1,16 @@
 #!/usr/bin/env python3
+"""Market Data Feeder — generates synthetic orders with realistic price dynamics.
+
+Price evolution uses regime-based simulation (trending / mean-reverting / volatile)
+so that technical indicators (SMA, EMA, MACD, RSI, Bollinger Bands) produce
+meaningful, non-random signals for the RL and LLM trading strategies.
+"""
+
 import sys
 sys.path.insert(0, "/app")
 
-import json, time, random, os, threading
+import json, math, time, random, os, threading
+from collections import deque
 
 from shared.config import Config
 from shared.kafka_utils import create_producer, create_consumer
@@ -15,6 +23,158 @@ _running = True
 _suspended = False
 
 
+# ── Price dynamics engine ─────────────────────────────────────────────────────
+
+class PriceDynamics:
+    """Per-symbol state for regime-driven price simulation."""
+
+    REGIMES = ("trending_up", "trending_down", "mean_revert", "volatile", "calm")
+
+    def __init__(self, symbol: str, start_price: float):
+        self.symbol = symbol
+        self.start_price = start_price
+        self.price = start_price
+        self.regime = random.choice(self.REGIMES)
+        self.regime_ticks = 0
+        self.regime_duration = random.randint(20, 60)
+        # Rolling history for indicator-aware order generation
+        self.history = deque(maxlen=60)
+        self.history.append(start_price)
+        # Momentum / mean-reversion state
+        self.momentum = 0.0
+        self.volatility = start_price * 0.005  # ~0.5% base volatility
+
+    def tick(self) -> float:
+        """Advance one tick and return the new mid price."""
+        tick_size = Config.TICK_SIZE
+        self.regime_ticks += 1
+
+        # Switch regime when duration expires
+        if self.regime_ticks >= self.regime_duration:
+            self._switch_regime()
+
+        # Compute drift + noise based on regime
+        if self.regime == "trending_up":
+            drift = self.volatility * random.uniform(0.2, 0.8)
+            noise = random.gauss(0, self.volatility * 0.5)
+            self.momentum = max(self.momentum * 0.95 + drift * 0.05, 0)
+        elif self.regime == "trending_down":
+            drift = -self.volatility * random.uniform(0.2, 0.8)
+            noise = random.gauss(0, self.volatility * 0.5)
+            self.momentum = min(self.momentum * 0.95 + drift * 0.05, 0)
+        elif self.regime == "mean_revert":
+            # Pull toward start price
+            pull = (self.start_price - self.price) * 0.03
+            drift = pull
+            noise = random.gauss(0, self.volatility * 0.3)
+            self.momentum *= 0.8
+        elif self.regime == "volatile":
+            drift = random.gauss(0, self.volatility * 0.5)
+            noise = random.gauss(0, self.volatility * 1.5)
+            self.momentum *= 0.9
+        else:  # calm
+            drift = random.gauss(0, self.volatility * 0.1)
+            noise = random.gauss(0, self.volatility * 0.2)
+            self.momentum *= 0.95
+
+        delta = drift + noise + self.momentum
+        # Quantize to tick size
+        delta = round(delta / tick_size) * tick_size
+        new_price = self.price + delta
+
+        # Floor at 1.00
+        new_price = max(1.00, round(new_price, 2))
+
+        # Adaptive volatility: expand in volatile regime, contract in calm
+        if self.regime == "volatile":
+            self.volatility = min(self.volatility * 1.002, self.price * 0.02)
+        elif self.regime == "calm":
+            self.volatility = max(self.volatility * 0.998, self.price * 0.002)
+
+        self.price = new_price
+        self.history.append(new_price)
+        return new_price
+
+    def _switch_regime(self):
+        """Transition to a new regime with weighted probabilities."""
+        # Bias: if price far from start, favor mean reversion
+        deviation = (self.price - self.start_price) / self.start_price
+        if abs(deviation) > 0.15:
+            weights = [0.05, 0.05, 0.50, 0.20, 0.20]
+        elif abs(deviation) > 0.08:
+            weights = [0.15, 0.15, 0.30, 0.20, 0.20]
+        else:
+            weights = [0.25, 0.25, 0.10, 0.15, 0.25]
+
+        self.regime = random.choices(self.REGIMES, weights=weights)[0]
+        self.regime_ticks = 0
+        self.regime_duration = random.randint(15, 50)
+
+    def sma(self, period: int) -> float:
+        h = list(self.history)
+        if len(h) < period:
+            return sum(h) / len(h)
+        return sum(h[-period:]) / period
+
+    def ema(self, period: int) -> float:
+        h = list(self.history)
+        alpha = 2.0 / (period + 1)
+        ema_val = h[0]
+        for p in h[1:]:
+            ema_val = alpha * p + (1 - alpha) * ema_val
+        return ema_val
+
+    def rsi(self, period: int = 14) -> float:
+        h = list(self.history)
+        if len(h) < 2:
+            return 50.0
+        gains, losses = [], []
+        for i in range(1, len(h)):
+            d = h[i] - h[i - 1]
+            gains.append(max(d, 0))
+            losses.append(max(-d, 0))
+        n = min(period, len(gains))
+        avg_gain = sum(gains[-n:]) / n if n > 0 else 0
+        avg_loss = sum(losses[-n:]) / n if n > 0 else 0.001
+        rs = avg_gain / avg_loss if avg_loss > 0 else 100
+        return 100 - 100 / (1 + rs)
+
+    def bollinger_position(self, period: int = 20) -> float:
+        """Returns 0..1 position within Bollinger Bands (0.5 = at SMA)."""
+        h = list(self.history)
+        n = min(period, len(h))
+        window = h[-n:]
+        mean = sum(window) / n
+        std = max((sum((x - mean) ** 2 for x in window) / n) ** 0.5, 0.01)
+        upper = mean + 2 * std
+        lower = mean - 2 * std
+        if upper == lower:
+            return 0.5
+        return max(0.0, min(1.0, (self.price - lower) / (upper - lower)))
+
+    def spread_and_depth(self) -> tuple:
+        """Compute adaptive spread and depth quantities based on regime."""
+        base_spread = 0.10
+        if self.regime == "volatile":
+            spread = base_spread * random.uniform(1.5, 2.5)
+            depth_qty_range = (30, 100)
+        elif self.regime == "calm":
+            spread = base_spread * random.uniform(0.6, 1.0)
+            depth_qty_range = (100, 300)
+        elif self.regime in ("trending_up", "trending_down"):
+            spread = base_spread * random.uniform(0.8, 1.5)
+            depth_qty_range = (50, 200)
+        else:
+            spread = base_spread
+            depth_qty_range = (80, 250)
+        return round(spread, 2), depth_qty_range
+
+
+_dynamics: dict[str, PriceDynamics] = {}
+
+
+# ── Securities I/O ────────────────────────────────────────────────────────────
+
 def load_securities():
     """Load securities from file: SYMBOL start_price current_price"""
     securities = {}
@@ -56,7 +216,8 @@ def make_order(symbol, side, price, qty):
     }
 
 
-def make_snapshot(symbol, best_bid, best_ask, bid_size, ask_size):
+def make_snapshot(symbol, best_bid, best_ask, bid_size, ask_size, dyn: PriceDynamics):
+    """Build snapshot with embedded technical indicators."""
     return {
         "symbol": symbol,
         "best_bid": round(best_bid, 2),
@@ -64,13 +225,23 @@ def make_snapshot(symbol, best_bid, best_ask, bid_size, ask_size):
         "bid_size": bid_size,
         "ask_size": ask_size,
         "timestamp": time.time(),
-        "source": "MDF"
+        "source": "MDF",
+        "indicators": {
+            "sma_5":  round(dyn.sma(5), 4),
+            "sma_20": round(dyn.sma(20), 4),
+            "ema_12": round(dyn.ema(12), 4),
+            "ema_26": round(dyn.ema(26), 4),
+            "macd":   round(dyn.ema(12) - dyn.ema(26), 4),
+            "rsi_14": round(dyn.rsi(14), 2),
+            "bb_pos": round(dyn.bollinger_position(20), 4),
+            "regime": dyn.regime,
+        },
     }
 
 
 def listen_control(ctrl_consumer):
     """Background thread: listen for start/stop/suspend/resume control messages."""
-    global _running, _suspended, _securities
+    global _running, _suspended, _securities, _dynamics
     print("[MDF] Control listener started")
     for msg in ctrl_consumer:
         action = (msg.value or {}).get("action")
@@ -83,6 +254,10 @@ def listen_control(ctrl_consumer):
                 new_secs = load_securities()
                 _securities.clear()
                 _securities.update(new_secs)
+                # Re-init dynamics for new session
+                _dynamics.clear()
+                for sym, vals in _securities.items():
+                    _dynamics[sym] = PriceDynamics(sym, vals["current"])
                 print(f"[MDF] START signal – reloaded securities: {list(_securities.keys())}")
             except Exception as e:
                 print(f"[MDF] Error reloading securities on start: {e}")
@@ -100,10 +275,11 @@ def listen_control(ctrl_consumer):
 if __name__ == "__main__":
     producer = create_producer(component_name="MDF")
 
-    # Load securities and snapshot start prices
+    # Load securities and init dynamics
     _securities = load_securities()
-    for sym in _securities:
-        _securities[sym]["start"] = _securities[sym]["current"]
+    for sym, vals in _securities.items():
+        vals["start"] = vals["current"]
+        _dynamics[sym] = PriceDynamics(sym, vals["current"])
     save_securities(_securities)
     print(f"[MDF] Loaded securities: {list(_securities.keys())}")
 
@@ -129,49 +305,72 @@ if __name__ == "__main__":
                 if not _running or _suspended:
                     break
 
-                mid = vals["current"]
-                half_spread = 0.10
+                dyn = _dynamics.get(sym)
+                if not dyn:
+                    continue
+
+                # Advance price via regime engine
+                mid = dyn.tick()
+                vals["current"] = mid
                 tick = Config.TICK_SIZE
+                spread, depth_qty_range = dyn.spread_and_depth()
+                half_spread = spread / 2
+
+                # ── Indicator-aware order bias ────────────────────────────
+                rsi = dyn.rsi(14)
+                bb_pos = dyn.bollinger_position(20)
+                macd = dyn.ema(12) - dyn.ema(26)
 
-                # Always place a resting BID and ASK to maintain book depth
+                # Bias aggressive side based on indicators
+                if rsi > 70 and bb_pos > 0.85:
+                    aggr_bias = "SELL"       # overbought → sellers step in
+                elif rsi < 30 and bb_pos < 0.15:
+                    aggr_bias = "BUY"        # oversold → buyers step in
+                elif macd > 0 and dyn.regime == "trending_up":
+                    aggr_bias = "BUY"
+                elif macd < 0 and dyn.regime == "trending_down":
+                    aggr_bias = "SELL"
+                else:
+                    aggr_bias = random.choice(["BUY", "SELL"])
+
+                # ── Place resting depth on both sides ─────────────────────
                 for depth_level in range(3):
                     offset = random.randint(1 + depth_level * 3, 3 + depth_level * 5) * tick
                     bid_price = round(mid - half_spread - offset, 2)
                     ask_price = round(mid + half_spread + offset, 2)
-                    bid_qty = random.choice([50, 100, 150, 200])
-                    ask_qty = random.choice([50, 100, 150, 200])
+                    bid_qty = random.randint(*depth_qty_range)
+                    ask_qty = random.randint(*depth_qty_range)
+
+                    # In trending regimes, skew depth: thicker on the passive side
+                    if dyn.regime == "trending_up" and depth_level == 0:
+                        bid_qty = int(bid_qty * 1.5)
+                    elif dyn.regime == "trending_down" and depth_level == 0:
+                        ask_qty = int(ask_qty * 1.5)
 
                     bid_order = make_order(sym, "BUY", bid_price, bid_qty)
                     ask_order = make_order(sym, "SELL", ask_price, ask_qty)
                     producer.send(Config.ORDERS_TOPIC, bid_order)
                     producer.send(Config.ORDERS_TOPIC, ask_order)
-                    print(f"[MDF] Depth: {sym} BID {bid_qty}@{bid_price:.2f}  ASK {ask_qty}@{ask_price:.2f}")
 
-                # Occasionally add an aggressive order to generate trades (20%)
-                if random.random() < 0.20:
-                    side = random.choice(["BUY", "SELL"])
+                # ── Aggressive orders to generate trades (20-35%) ─────────
+                aggr_prob = 0.35 if dyn.regime == "volatile" else 0.20
+                if random.random() < aggr_prob:
+                    side = aggr_bias
                     if side == "BUY":
                         price = round(mid + half_spread + random.randint(1, 3) * tick, 2)
                     else:
                         price = round(mid - half_spread - random.randint(1, 3) * tick, 2)
-                    qty = random.choice([50, 100, 150])
+                    qty = random.randint(*depth_qty_range)
                     aggr_order = make_order(sym, side, price, qty)
                     producer.send(Config.ORDERS_TOPIC, aggr_order)
-                    print(f"[MDF] Aggr:  {sym} {side} {qty}@{price:.2f}")
-
-                # Simulate small price drift (10% chance, max 2 ticks)
-                if random.random() < 0.10:
-                    drift = random.choice([-2, -1, 1, 2]) * tick
-                    new_price = vals["current"] + drift
-                    if new_price >= 1.00:
-                        vals["current"] = round(new_price, 2)
-                        save_securities(_securities)
 
+                # ── Persist and publish snapshot with indicators ───────────
+                save_securities(_securities)
                 best_bid = round(mid - half_spread, 2)
                 best_ask = round(mid + half_spread, 2)
-                bid_size = random.choice([100, 200, 300])
-                ask_size = random.choice([100, 200, 300])
-                snap = make_snapshot(sym, best_bid, best_ask, bid_size, ask_size)
+                bid_size = random.randint(100, 400)
+                ask_size = random.randint(100, 400)
+                snap = make_snapshot(sym, best_bid, best_ask, bid_size, ask_size, dyn)
                 producer.send(Config.SNAPSHOTS_TOPIC, snap)
 
                 time.sleep(ORDER_INTERVAL)