87 lines
3.3 KiB
Python
87 lines
3.3 KiB
Python
|
#!/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()
|