408 lines
16 KiB
Python
408 lines
16 KiB
Python
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
"""
|
|
router/main.py — Centaur Quant Architecture | Central Gateway (Python Router)
|
|
|
|
Role
|
|
----
|
|
Central TCP/IP gateway for the Universal Communication Bridge. The MQL5
|
|
Executor (CentaurQuant.mq5) streams SDP JSON frames into this process on
|
|
127.0.0.1:5555. This router:
|
|
|
|
1. Accepts persistent client connections (one thread per connection).
|
|
2. Reassembles frames terminated by '\n' or '\r\n' from raw TCP bytes.
|
|
3. Parses each frame STRICTLY with the stdlib `json` module.
|
|
4. Dispatches on the SDP "action_type" key:
|
|
Heartbeat / Tick_Harvest -> lightweight console log
|
|
Setup_Detected -> LIVE LLM evaluation (MarketAnalyzer) + reply
|
|
Trade_Opened / Closed -> prominent log + SQLite telemetry persist
|
|
5. Replies to Setup_Detected with the AI Advisory envelope:
|
|
{"algorithmic_confidence_score": <float>}\n
|
|
|
|
The reply is advisory ONLY (Anti-Veto Principle): the MT5 side may still
|
|
execute regardless of the score. MarketAnalyzer returns a neutral 40.0 on
|
|
any LLM failure/timeout, which triggers the MT5 quarter-risk path.
|
|
Trade_Opened/Trade_Closed are persisted by TelemetryDB (Track B), closing
|
|
the score -> execution -> outcome feedback loop for future calibration.
|
|
|
|
Robustness
|
|
----------
|
|
- Per-connection threads are daemonized; the server never dies with a client.
|
|
- ConnectionResetError / BrokenPipeError / empty reads are handled as a
|
|
clean disconnect; a reconnect is simply a new accept().
|
|
- Frames and the reassembly buffer are size-bounded to resist floods.
|
|
|
|
Standard library only: socket, threading, json, logging, argparse.
|
|
"""
|
|
|
|
import argparse
|
|
import json
|
|
import logging
|
|
import os
|
|
import socket
|
|
import sys
|
|
import threading
|
|
from typing import Any, Dict, List, Optional
|
|
|
|
# --- Track A: live LLM evaluation engine (Python/models/analyzer.py) -------
|
|
# Make the sibling packages importable whether run as a script or as a
|
|
# module: python router/main.py | python -m router.main
|
|
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
|
from models.analyzer import MarketAnalyzer # noqa: E402
|
|
from database.telemetry import TelemetryDB # noqa: E402
|
|
|
|
# --------------------------------------------------------------------------
|
|
# Configuration
|
|
# --------------------------------------------------------------------------
|
|
HOST = "127.0.0.1" # default listen address (MT5 side connects here)
|
|
PORT = 5555 # default listen port (matches CentaurQuant.mq5 input)
|
|
BACKLOG = 5 # pending-connection queue depth
|
|
RECV_BUFFER_SIZE = 4096 # bytes pulled per recv() call
|
|
MAX_FRAME_SIZE = 1_048_576 # 1 MB hard cap per logical frame (flood guard)
|
|
ENCODING = "utf-8" # SDP frames are UTF-8 encoded
|
|
|
|
# --------------------------------------------------------------------------
|
|
# AI evaluation engine (Track A)
|
|
# Provider/model from env: PROVIDER=auto|openai|google, MODEL, *_API_KEY.
|
|
# Anti-Veto: analyze() returns 40.0 (quarter-risk trigger) on any failure.
|
|
# --------------------------------------------------------------------------
|
|
ANALYZER = MarketAnalyzer()
|
|
|
|
# --------------------------------------------------------------------------
|
|
# Telemetry persistence (Track B) — SQLite feedback-loop store.
|
|
# DB path overridable via env: CENTAUR_DB_PATH (default: ./centaur_telemetry.db)
|
|
# --------------------------------------------------------------------------
|
|
TELEMETRY = TelemetryDB(os.environ.get("CENTAUR_DB_PATH", "centaur_telemetry.db"))
|
|
|
|
# --------------------------------------------------------------------------
|
|
# Logging
|
|
# --------------------------------------------------------------------------
|
|
def configure_logging() -> None:
|
|
"""Console logging with timestamps; INFO shows the full SDP telemetry."""
|
|
logging.basicConfig(
|
|
level=logging.INFO,
|
|
format="%(asctime)s | %(levelname)-8s | %(message)s",
|
|
datefmt="%Y-%m-%d %H:%M:%S",
|
|
)
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# Frame reassembly (newline / CRLF framed protocol)
|
|
# --------------------------------------------------------------------------
|
|
class FrameReader:
|
|
"""
|
|
Incrementally reassembles '\n'-terminated text frames from raw bytes.
|
|
|
|
The MT5 transport (CSocketClient) writes frames terminated by CRLF
|
|
("\r\n") and the receiver tolerates bare "\n" — so we split on "\n"
|
|
and strip one trailing "\r" when present. Multiple frames may arrive
|
|
inside a single recv(); the remainder is kept buffered across reads.
|
|
"""
|
|
|
|
def __init__(self) -> None:
|
|
self._buffer = b""
|
|
|
|
def feed(self, data: bytes) -> List[str]:
|
|
"""Append raw bytes and return every COMPLETE frame they contain."""
|
|
self._buffer += data
|
|
frames: List[str] = []
|
|
|
|
while True:
|
|
idx = self._buffer.find(b"\n")
|
|
if idx == -1:
|
|
break # no complete frame yet — keep buffering
|
|
|
|
raw = self._buffer[:idx] # frame body (terminator removed)
|
|
self._buffer = self._buffer[idx + 1:] # consume terminator too
|
|
|
|
if raw.endswith(b"\r"):
|
|
raw = raw[:-1] # CRLF tolerance
|
|
|
|
if len(raw) > MAX_FRAME_SIZE:
|
|
logging.warning("Frame exceeds %d bytes — dropped.", MAX_FRAME_SIZE)
|
|
continue
|
|
|
|
try:
|
|
frames.append(raw.decode(ENCODING, errors="replace"))
|
|
except UnicodeDecodeError as exc: # pragma: no cover - defensive
|
|
logging.warning("Undecodable frame dropped: %s", exc)
|
|
|
|
# Flood guard: never let an unframed stream grow the buffer unbounded.
|
|
if len(self._buffer) > MAX_FRAME_SIZE:
|
|
logging.warning(
|
|
"Reassembly buffer exceeded %d bytes without a terminator — cleared.",
|
|
MAX_FRAME_SIZE,
|
|
)
|
|
self._buffer = b""
|
|
|
|
return frames
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# Strict SDP parsing
|
|
# --------------------------------------------------------------------------
|
|
def parse_sdp_frame(frame: str) -> Optional[Dict[str, Any]]:
|
|
"""
|
|
Strictly parse one SDP frame.
|
|
|
|
Returns a dict only when the frame is valid JSON *and* an object *and*
|
|
carries a string "action_type"; otherwise logs and returns None.
|
|
"""
|
|
try:
|
|
data = json.loads(frame)
|
|
except json.JSONDecodeError as exc:
|
|
logging.warning("Malformed JSON dropped: %s | frame=%.120r", exc, frame)
|
|
return None
|
|
|
|
if not isinstance(data, dict):
|
|
logging.warning("Non-object JSON dropped: %r", data)
|
|
return None
|
|
|
|
if not isinstance(data.get("action_type"), str):
|
|
logging.warning("Frame missing string 'action_type': %r", data)
|
|
return None
|
|
|
|
return data
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# Safe transport write (client may vanish between read and reply)
|
|
# --------------------------------------------------------------------------
|
|
def safe_send(conn: socket.socket, text: str) -> bool:
|
|
"""Best-effort frame write; never raises on a dead client."""
|
|
try:
|
|
conn.sendall(text.encode(ENCODING))
|
|
return True
|
|
except (ConnectionResetError, BrokenPipeError, OSError) as exc:
|
|
logging.warning("Send failed (client gone?): %s", exc)
|
|
return False
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# Action handlers
|
|
# --------------------------------------------------------------------------
|
|
def handle_heartbeat(msg: Dict[str, Any]) -> None:
|
|
"""Liveness signal — lightweight one-line log."""
|
|
logging.info(
|
|
"[Heartbeat] symbol=%s timeframe=%s",
|
|
msg.get("symbol", "?"),
|
|
msg.get("timeframe", "?"),
|
|
)
|
|
|
|
|
|
def handle_tick_harvest(msg: Dict[str, Any]) -> None:
|
|
"""One harvested tick — compact telemetry line."""
|
|
p = msg.get("payload") or {}
|
|
logging.info(
|
|
"[Tick_Harvest] %s %s bid=%s ask=%s spread_pts=%s tick_vol=%s",
|
|
msg.get("symbol", "?"),
|
|
msg.get("timeframe", "?"),
|
|
p.get("bid", "?"),
|
|
p.get("ask", "?"),
|
|
p.get("spread_points", "?"),
|
|
p.get("tick_volume", "?"),
|
|
)
|
|
|
|
|
|
def handle_setup_detected(msg: Dict[str, Any], conn: socket.socket) -> None:
|
|
"""
|
|
Setup proposal from the Executor.
|
|
|
|
Extracts symbol / setup_type / historical_context, runs the LIVE LLM
|
|
evaluation (MarketAnalyzer), and INSTANTLY replies with the AI Advisory
|
|
envelope:
|
|
{"algorithmic_confidence_score": <float>}\n
|
|
"""
|
|
p = msg.get("payload") or {}
|
|
symbol = msg.get("symbol", "?")
|
|
setup_type = p.get("setup_type", "?")
|
|
|
|
hist = msg.get("historical_context")
|
|
swing_count = len(hist) if isinstance(hist, list) else 0
|
|
|
|
logging.info(
|
|
"[Setup_Detected] %s | setup=%s | entry=%s sl=%s tp=%s | ctx_swings=%d",
|
|
symbol,
|
|
setup_type,
|
|
p.get("entry", "?"),
|
|
p.get("sl", "?"),
|
|
p.get("tp", "?"),
|
|
swing_count,
|
|
)
|
|
|
|
# --- live LLM evaluation (Anti-Veto: neutral 40.0 on failure/timeout) ---
|
|
result = ANALYZER.analyze(msg)
|
|
score = result["score"]
|
|
if result.get("used_fallback"):
|
|
logging.warning("[AI_Advisory] %s | fallback score %.2f used — %s",
|
|
symbol, score, result.get("fallback_reason", ""))
|
|
else:
|
|
logging.info("[AI_Advisory] %s | score=%.2f | reason='%s' | provider=%s | %d ms",
|
|
symbol, score, result.get("reason", ""),
|
|
result.get("provider", "?"), result.get("latency_ms", 0))
|
|
|
|
# --- AI Advisory envelope, newline-terminated (MT5 frames on '\n') ---
|
|
reply = json.dumps({"algorithmic_confidence_score": score}) + "\n"
|
|
if safe_send(conn, reply):
|
|
logging.info("[AI_Advisory] %s | score=%.2f -> sent to MT5", symbol, score)
|
|
else:
|
|
logging.warning(
|
|
"[AI_Advisory] %s | score=%.2f NOT delivered (client unreachable)",
|
|
symbol,
|
|
score,
|
|
)
|
|
|
|
|
|
def handle_trade_opened(msg: Dict[str, Any]) -> None:
|
|
"""Position opened — prominent execution record + SQLite persist."""
|
|
p = msg.get("payload") or {}
|
|
logging.info(
|
|
">>> [Trade_Opened] %s %s | ticket=%s %s lot=%s entry=%s sl=%s tp=%s | ai_score=%s",
|
|
msg.get("symbol", "?"),
|
|
msg.get("timeframe", "?"),
|
|
p.get("ticket", "?"),
|
|
p.get("direction", "?"),
|
|
p.get("lot", "?"),
|
|
p.get("entry_price", "?"),
|
|
p.get("sl", "?"),
|
|
p.get("tp", "?"),
|
|
msg.get("algorithmic_confidence_score", "?"),
|
|
)
|
|
# --- persist the execution record (Track B) ---
|
|
TELEMETRY.log_trade_opened(msg)
|
|
|
|
|
|
def handle_trade_closed(msg: Dict[str, Any]) -> None:
|
|
"""Position closed — the feedback-loop training signal + SQLite update."""
|
|
p = msg.get("payload") or {}
|
|
logging.info(
|
|
">>> [Trade_Closed] %s | ticket=%s profit=%s r_multiple=%s initial_ai_score=%s",
|
|
msg.get("symbol", "?"),
|
|
p.get("ticket", "?"),
|
|
p.get("profit", "?"),
|
|
p.get("r_multiple", "?"),
|
|
p.get("initial_ai_score", "?"),
|
|
)
|
|
# --- persist the closed outcome (Track B) ---
|
|
TELEMETRY.log_trade_closed(msg)
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# Dispatcher
|
|
# --------------------------------------------------------------------------
|
|
# Setup_Detected is handled separately (it needs the socket to reply).
|
|
_ACTIONS: Dict[str, Any] = {
|
|
"Heartbeat": handle_heartbeat,
|
|
"Tick_Harvest": handle_tick_harvest,
|
|
"Trade_Opened": handle_trade_opened,
|
|
"Trade_Closed": handle_trade_closed,
|
|
}
|
|
|
|
|
|
def dispatch(msg: Dict[str, Any], conn: socket.socket) -> None:
|
|
"""Route one parsed SDP frame by its action_type."""
|
|
action = msg.get("action_type")
|
|
|
|
if action == "Setup_Detected":
|
|
handle_setup_detected(msg, conn)
|
|
return
|
|
|
|
handler = _ACTIONS.get(action)
|
|
if handler is None:
|
|
logging.warning("[Router] unknown action_type=%r — frame ignored", action)
|
|
return
|
|
|
|
try:
|
|
handler(msg)
|
|
except Exception: # a broken handler must never kill the connection thread
|
|
logging.exception("[Router] handler '%s' raised — frame ignored", action)
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# Per-connection worker
|
|
# --------------------------------------------------------------------------
|
|
def handle_client(conn: socket.socket, addr: tuple) -> None:
|
|
"""
|
|
Serve one persistent MT5 connection until it disconnects.
|
|
|
|
Reads raw bytes, reassembles frames, parses and dispatches each one.
|
|
A reconnect from the MT5 side simply spawns a fresh thread via accept().
|
|
"""
|
|
host, port = addr[0], addr[1]
|
|
logging.info("[Gateway] client connected: %s:%d", host, port)
|
|
|
|
reader = FrameReader()
|
|
try:
|
|
while True:
|
|
data = conn.recv(RECV_BUFFER_SIZE)
|
|
if not data:
|
|
# Peer performed an orderly close — empty byte read.
|
|
logging.info("[Gateway] %s:%d closed connection (empty read)", host, port)
|
|
break
|
|
|
|
for frame in reader.feed(data):
|
|
msg = parse_sdp_frame(frame)
|
|
if msg is not None:
|
|
dispatch(msg, conn)
|
|
|
|
except ConnectionResetError as exc:
|
|
# MT5 terminal killed the socket (e.g., chart closed mid-session).
|
|
logging.warning("[Gateway] %s:%d connection reset: %s", host, port, exc)
|
|
except (BrokenPipeError, OSError) as exc:
|
|
logging.warning("[Gateway] %s:%d socket error: %s", host, port, exc)
|
|
except Exception:
|
|
# Last-resort guard: log and drop the connection, keep the server up.
|
|
logging.exception("[Gateway] %s:%d unexpected error — connection closed", host, port)
|
|
finally:
|
|
try:
|
|
conn.shutdown(socket.SHUT_RDWR)
|
|
except OSError:
|
|
pass # socket may already be closed by the peer
|
|
conn.close()
|
|
logging.info("[Gateway] %s:%d disconnected", host, port)
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# Server entry point
|
|
# --------------------------------------------------------------------------
|
|
def run_server(host: str, port: int) -> None:
|
|
"""Bind, listen, and accept client connections forever (Ctrl+C to stop)."""
|
|
server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
|
server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) # fast rebind
|
|
server.bind((host, port))
|
|
server.listen(BACKLOG)
|
|
|
|
logging.info("=" * 72)
|
|
logging.info("Centaur Gateway listening on %s:%d", host, port)
|
|
logging.info("LLM backend: %s",
|
|
ANALYZER._backend.name if ANALYZER._backend else "NONE (fallback 40.0)")
|
|
logging.info("Telemetry DB: %s", TELEMETRY.path)
|
|
logging.info("Press Ctrl+C to stop.")
|
|
logging.info("=" * 72)
|
|
|
|
try:
|
|
while True:
|
|
conn, addr = server.accept()
|
|
# daemon thread: dies with the process; never blocks shutdown
|
|
thread = threading.Thread(target=handle_client, args=(conn, addr), daemon=True)
|
|
thread.start()
|
|
except KeyboardInterrupt:
|
|
logging.info("[Gateway] Ctrl+C received — shutting down.")
|
|
finally:
|
|
server.close()
|
|
TELEMETRY.close()
|
|
logging.info("[Gateway] server socket closed, telemetry flushed. Bye.")
|
|
|
|
|
|
def main() -> None:
|
|
parser = argparse.ArgumentParser(description="Centaur Quant — Central Gateway")
|
|
parser.add_argument("--host", default=HOST, help="listen address (default: %(default)s)")
|
|
parser.add_argument("--port", type=int, default=PORT, help="listen port (default: %(default)s)")
|
|
args = parser.parse_args()
|
|
|
|
configure_logging()
|
|
run_server(args.host, args.port)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|