""" Mock WebSocket server for MA+IV system development. Pushes fake market data so the frontend can be developed without real API credentials. All message formats match the schema defined in CLAUDE.md. Endpoint: ws://localhost:8765 (no Token auth — dev only) Push schedule: on connect kline_init 120 daily bars per symbol every 1 s quote price ±0.3 % random walk every 30 s iv small IV / IVP drift every 60 s signal 1-3 random symbols, score update every ~5 m alert one random high-score symbol """ import asyncio import json import logging import random import time from datetime import datetime, timedelta, timezone import websockets logging.basicConfig( level=logging.INFO, format="%(asctime)s %(levelname)-5s %(message)s", ) log = logging.getLogger(__name__) SYMBOLS = ["AAPL", "MSFT", "NVDA", "TSLA", "META", "AMZN", "GOOGL", "AMD", "NFLX", "SMCI"] # All connected WebSocket clients _clients: set = set() # Shared mutable state for all symbols STATE: dict = {} # --------------------------------------------------------------------------- # State initialisation # --------------------------------------------------------------------------- def _init_state() -> dict: state = {} for sym in SYMBOLS: call_iv = round(random.uniform(0.15, 0.60), 3) put_iv = round(call_iv + random.uniform(0.00, 0.08), 3) state[sym] = { "price": round(random.uniform(100, 900), 2), "change_pct": round(random.uniform(-3.0, 3.0), 2), "volume": random.randint(500_000, 5_000_000), "score": random.randint(30, 95), "convergence_days": random.randint(0, 45), "bandwidth_pct": round(random.uniform(5.0, 40.0), 1), "ivp": random.randint(5, 80), "call_iv": call_iv, "put_iv": put_iv, "skew": round((put_iv - call_iv) * 100, 1), "pcr": round(random.uniform(0.6, 1.4), 2), "hv": round(call_iv * 0.85, 3), } return state # --------------------------------------------------------------------------- # K-line generation # --------------------------------------------------------------------------- def _trading_timestamps(n: int) -> list[int]: """Return n Unix timestamps (16:00 ET) for the last n weekdays.""" now = datetime.now(timezone.utc) result: list[int] = [] day = now - timedelta(days=int(n * 1.5) + 5) while len(result) < n: if day.weekday() < 5: # Mon-Fri result.append(int(day.replace(hour=21, minute=0, second=0, microsecond=0).timestamp())) day += timedelta(days=1) return result[:n] def _generate_bars(base_price: float, n: int = 120) -> list[dict]: """Build n OHLCV daily bars as a realistic random walk.""" # Walk backwards from base_price then reverse so bars are chronological closes = [base_price] for _ in range(n - 1): closes.append(max(5.0, closes[-1] * (1 + random.uniform(-0.025, 0.025)))) closes.reverse() timestamps = _trading_timestamps(n) bars = [] for ts, close in zip(timestamps, closes): open_ = round(close * (1 + random.uniform(-0.012, 0.012)), 2) high = round(max(open_, close) * (1 + random.uniform(0.001, 0.018)), 2) low = round(min(open_, close) * (1 - random.uniform(0.001, 0.018)), 2) volume = random.randint(300_000, 6_000_000) bars.append({"t": ts, "o": open_, "h": high, "l": low, "c": round(close, 2), "v": volume}) return bars # --------------------------------------------------------------------------- # Send helpers # --------------------------------------------------------------------------- def _ts() -> int: return int(time.time()) async def _send(ws, data: dict) -> bool: """Send JSON to one client. Returns False if the connection is dead.""" try: payload = json.dumps(data) await ws.send(payload) log.info("[→ %s] %s", getattr(ws, "remote_address", "?"), payload[:160]) return True except Exception: return False async def _broadcast(data: dict): """Send to all clients; prune dead ones.""" dead = {ws for ws in list(_clients) if not await _send(ws, data)} _clients.difference_update(dead) # --------------------------------------------------------------------------- # On-connect burst # --------------------------------------------------------------------------- async def _push_kline_init(ws): for sym in SYMBOLS: bars = _generate_bars(STATE[sym]["price"]) ok = await _send(ws, {"type": "kline_init", "symbol": sym, "bars": bars}) if not ok: return # --------------------------------------------------------------------------- # Broadcast loops # --------------------------------------------------------------------------- async def _loop_quote(): """Every 1 s: push quote for every symbol.""" while True: await asyncio.sleep(1) if not _clients: continue for sym in SYMBOLS: d = STATE[sym] d["price"] = round(max(1.0, d["price"] * (1 + random.uniform(-0.003, 0.003))), 2) d["change_pct"] = round(max(-15.0, min(15.0, d["change_pct"] + random.uniform(-0.08, 0.08))), 2) d["volume"] += random.randint(1_000, 80_000) await _broadcast({ "type": "quote", "symbol": sym, "price": d["price"], "change_pct": d["change_pct"], "volume": d["volume"], "ts": _ts(), }) async def _loop_iv(): """Every 30 s: push iv snapshot for every symbol.""" while True: await asyncio.sleep(30) if not _clients: continue for sym in SYMBOLS: d = STATE[sym] d["call_iv"] = round(max(0.05, d["call_iv"] + random.uniform(-0.012, 0.012)), 3) d["put_iv"] = round(max(0.05, d["put_iv"] + random.uniform(-0.012, 0.012)), 3) d["ivp"] = max(1, min(99, d["ivp"] + random.randint(-4, 4))) d["skew"] = round((d["put_iv"] - d["call_iv"]) * 100, 1) d["pcr"] = round(max(0.3, min(2.5, d["pcr"] + random.uniform(-0.06, 0.06))), 2) d["hv"] = round(d["call_iv"] * 0.85, 3) await _broadcast({ "type": "iv", "symbol": sym, "call_iv": d["call_iv"], "put_iv": d["put_iv"], "ivp": d["ivp"], "skew": d["skew"], "pcr": d["pcr"], "hv": d["hv"], "ts": _ts(), }) async def _loop_signal(): """Every 60 s: update score for 1-3 random symbols.""" while True: await asyncio.sleep(60) if not _clients: continue for sym in random.sample(SYMBOLS, k=random.randint(1, 3)): d = STATE[sym] d["score"] = max(0, min(100, d["score"] + random.randint(-12, 12))) d["convergence_days"] = max(0, d["convergence_days"] + random.randint(-2, 4)) d["bandwidth_pct"] = round(max(1.0, d["bandwidth_pct"] + random.uniform(-3.0, 3.0)), 1) await _broadcast({ "type": "signal", "symbol": sym, "score": d["score"], "convergence_days": d["convergence_days"], "bandwidth_pct": d["bandwidth_pct"], "ts": _ts(), }) _ALERT_REASONS = [ "均线粘合{days}天，IVP低位{ivp}%，Skew收窄", "MA密集排列{days}天，隐波{ivp}%历史低位，布林带收口", "均线粘合{days}天，IVP处于{ivp}%分位，突破概率上升", ] async def _loop_alert(): """Every ~300 s: push one alert for a random high-score symbol.""" while True: await asyncio.sleep(random.uniform(240, 360)) if not _clients: continue sym = random.choice(SYMBOLS) d = STATE[sym] reason = random.choice(_ALERT_REASONS).format( days=d["convergence_days"], ivp=d["ivp"] ) await _broadcast({ "type": "alert", "symbol": sym, "score": d["score"], "reason": reason, "ts": _ts(), }) # --------------------------------------------------------------------------- # Connection handler # --------------------------------------------------------------------------- async def _handler(ws): addr = ws.remote_address log.info("Client connected: %s", addr) _clients.add(ws) try: await _push_kline_init(ws) async for _ in ws: pass # server does not expect client messages in this version except websockets.exceptions.ConnectionClosed: pass finally: _clients.discard(ws) log.info("Client disconnected: %s", addr) # --------------------------------------------------------------------------- # Entry point # --------------------------------------------------------------------------- async def main(): global STATE STATE = _init_state() log.info("Mock server running on ws://localhost:8765") async with websockets.serve(_handler, "127.0.0.1", 8765): await asyncio.gather( _loop_quote(), _loop_iv(), _loop_signal(), _loop_alert(), ) if __name__ == "__main__": asyncio.run(main())