Centaur_Quant_Architecture/Python/ingest_telemetry.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
ingest_telemetry.py — File-based telemetry fallback ingester (Track B).

Reads MQL5\\Files\\telemetry.jsonl (written by CentaurQuant.mq5 when the
TCP transport is unavailable — see AppendTelemetryFile in the EA) and
imports SDP Trade_Opened / Trade_Closed frames into centaur_telemetry.db
via TelemetryDB — the same schema/logic the router uses.

Safe to run repeatedly: TelemetryDB dedupes on the PRIMARY KEY ticket
(INSERT OR IGNORE for opens, UPDATE for closes).

Usage:
    python ingest_telemetry.py [--jsonl PATH] [--db PATH] [--keep]
"""

import argparse
import json
import os
import sys

sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from database.telemetry import TelemetryDB  # noqa: E402

DEFAULT_JSONL = r"D:\TradingTerminal\MetaTrader 5\MQL5\Files\telemetry.jsonl"
DEFAULT_DB = os.path.join(os.path.dirname(os.path.abspath(__file__)),
                          "centaur_telemetry.db")


def main() -> None:
    ap = argparse.ArgumentParser(description="Centaur file-telemetry ingester")
    ap.add_argument("--jsonl", default=DEFAULT_JSONL,
                    help="path to telemetry.jsonl (default: %(default)s)")
    ap.add_argument("--db", default=DEFAULT_DB,
                    help="SQLite telemetry DB path (default: %(default)s)")
    ap.add_argument("--keep", action="store_true",
                    help="do NOT truncate the JSONL after a successful ingest")
    args = ap.parse_args()

    if not os.path.exists(args.jsonl):
        print(f"ingest_telemetry: no telemetry file at {args.jsonl} — nothing to do.")
        return

    db = TelemetryDB(args.db)
    opened = closed = skipped = 0
    try:
        with open(args.jsonl, "r", encoding="utf-8", errors="replace") as f:
            for line in f:
                line = line.strip()
                if not line:
                    continue
                try:
                    msg = json.loads(line)
                except json.JSONDecodeError:
                    skipped += 1
                    continue
                action = msg.get("action_type")
                try:
                    if action == "Trade_Opened":
                        if db.log_trade_opened(msg):
                            opened += 1
                    elif action == "Trade_Closed":
                        if db.log_trade_closed(msg):
                            closed += 1
                    else:
                        skipped += 1   # heartbeat/tick frames are not persisted
                except Exception as exc:  # a bad frame must not abort the batch
                    skipped += 1
                    print(f"ingest_telemetry: frame error: {exc!r}")
    finally:
        db.close()

    print(f"ingest_telemetry: opened={opened} closed={closed} skipped={skipped}")

    # Consume the file only when something was processed; keeps the EA's
    # concurrent appends from being lost in a race with an empty truncate.
    if not args.keep and (opened + closed + skipped) > 0:
        try:
            open(args.jsonl, "w", encoding="utf-8").close()
            print(f"ingest_telemetry: {args.jsonl} truncated (consumed).")
        except OSError as exc:
            print(f"ingest_telemetry: WARNING could not truncate: {exc}")


if __name__ == "__main__":
    main()