"""PropMemo Tracker for TradeLocker — prop-memo.com 収支トラッカー連携スクリプト TradeLocker 口座の残高・有効証拠金を定期取得し、prop-memo.com の収支トラッカーAPIへ送信する。 MT4/MT5版EA・cTrader版cBotと同じAPI（同じAPIキー）で動作する。 依存: pip install tradelocker requests python-dotenv 環境変数（.env対応 — TradeLocker botと同じ .env に相乗り可能）: TL_ENV=demo|live 接続先（既定 demo） TL_USERNAME / TL_PASSWORD / TL_SERVER TradeLockerログイン情報（必須） TL_ACCOUNT_ID=0 0=自動（最初の口座） TL_ACC_NUM=0 0=自動 PROPMEMO_API_KEY prop-memo.com 収支ページで発行したAPIキー（必須） PROPMEMO_API_BASE 既定 https://prop-memo.com PROPMEMO_FIRM_SLUG ファーム自動紐づけ用slug（既定 superfunded） PROPMEMO_INTERVAL=60 送信間隔秒（常駐時） PROPMEMO_STATE_FILE=propmemo_state.json 日次損益計算用の状態ファイル 使い方: python propmemo_tl_tracker.py # 常駐ループ（ローカル/VM向け） python propmemo_tl_tracker.py --once # 1回送信して終了（cron / Cloud Run Job向け） """ from __future__ import annotations import json import os import sys import time from datetime import datetime, timezone from pathlib import Path from zoneinfo import ZoneInfo import requests try: from dotenv import load_dotenv load_dotenv() except ImportError: pass from tradelocker import TLAPI JST = ZoneInfo("Asia/Tokyo") RECONNECT_AFTER_FAILURES = 3 DEMO_URL = "https://demo.tradelocker.com" LIVE_URL = "https://live.tradelocker.com" def _env(name: str, default: str = "") -> str: v = os.getenv(name) return v.strip() if v is not None else default def _pick(state: dict, *names: str) -> float | None: """口座状態dictから候補キーの最初に見つかった数値を返す（キーはサーバー定義のため防御的に）。""" for n in names: if n in state and state[n] is not None: try: return float(state[n]) except (TypeError, ValueError): continue return None class PropMemoClient: def __init__(self) -> None: self.base = _env("PROPMEMO_API_BASE", "https://prop-memo.com").rstrip("/") self.api_key = _env("PROPMEMO_API_KEY") self.firm_slug = _env("PROPMEMO_FIRM_SLUG", "superfunded") if not self.api_key: raise SystemExit("PROPMEMO_API_KEY が未設定です。prop-memo.com の収支ページでAPIキーを発行してください。") def _post(self, path: str, payload: dict) -> dict: r = requests.post(f"{self.base}{path}", json=payload, timeout=15) r.raise_for_status() return r.json() def register(self, account_number: int, server: str, currency: str, balance: float, equity: float, is_demo: bool) -> dict: return self._post("/api/ea/register", { "api_key": self.api_key, "account_number": account_number, "server": server, "broker_name": "TradeLocker", "currency": currency, "balance": balance, "equity": equity, "is_demo": is_demo, "detected_firm_slug": self.firm_slug, }) def snapshot(self, account_number: int, server: str, balance: float, equity: float) -> dict: return self._post("/api/ea/snapshot", { "api_key": self.api_key, "account_number": account_number, "server": server, "balance": balance, "equity": equity, "recorded_at": datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%S"), "detected_firm_slug": self.firm_slug, }) def daily_pnl(self, account_number: int, server: str, date_str: str, pnl: float) -> dict: return self._post("/api/ea/daily-pnl", { "api_key": self.api_key, "account_number": account_number, "server": server, "daily_pnl": [{"date": date_str, "pnl": round(pnl, 2)}], }) class Tracker: def __init__(self) -> None: self.env_name = (_env("TL_ENV", "demo") or "demo").lower() self.server = _env("TL_SERVER") if not (_env("TL_USERNAME") and _env("TL_PASSWORD") and self.server): raise SystemExit("TL_USERNAME / TL_PASSWORD / TL_SERVER が未設定です（.env.example参照）。") self.pm = PropMemoClient() self.state_path = Path(_env("PROPMEMO_STATE_FILE", "propmemo_state.json")) self.tl: TLAPI | None = None self.account_number = 0 self.currency = "USD" # ---- TradeLocker ---- def connect(self) -> None: self.tl = TLAPI( environment=LIVE_URL if self.env_name == "live" else DEMO_URL, username=_env("TL_USERNAME"), password=_env("TL_PASSWORD"), server=self.server, account_id=int(_env("TL_ACCOUNT_ID", "0") or 0), acc_num=int(_env("TL_ACC_NUM", "0") or 0), log_level="warning", ) # TLAPI 初期化後は口座が解決済み（0=自動指定でも実IDが入る） self.account_number = int(self.tl.account_id) try: accounts = self.tl.get_all_accounts() row = accounts[accounts["id"].astype(int) == self.account_number] if not row.empty and "currency" in row.columns: self.currency = str(row.iloc[0]["currency"]) or "USD" except Exception as e: # noqa: BLE001 print(f"[warn] 口座通貨の取得に失敗（USDで継続）: {e}") def balance_equity(self) -> tuple[float, float]: state = self.tl.get_account_state() balance = _pick(state, "balance") if balance is None: raise RuntimeError(f"口座状態にbalanceが見つかりません。keys={list(state.keys())}") # equity相当: projectedBalance（残高+含み損益）が無ければ含み損益を加算、それも無ければ残高 equity = _pick(state, "projectedBalance") if equity is None: open_pnl = _pick(state, "openNetPnL", "openGrossPnL") equity = balance + open_pnl if open_pnl is not None else balance return balance, equity # ---- 日次損益（JST日付・日初残高との差分）---- def _load_state(self) -> dict: try: return json.loads(self.state_path.read_text(encoding="utf-8")) except (FileNotFoundError, json.JSONDecodeError): return {} def _save_state(self, state: dict) -> None: tmp = self.state_path.with_suffix(".tmp") tmp.write_text(json.dumps(state, ensure_ascii=False), encoding="utf-8") tmp.replace(self.state_path) def today_pnl(self, balance: float) -> tuple[str, float]: today = datetime.now(JST).strftime("%Y-%m-%d") state = self._load_state() if state.get("day") != today: state = {"day": today, "day_start_balance": balance} self._save_state(state) return today, balance - float(state.get("day_start_balance", balance)) # ---- 1回分の送信 ---- def tick(self) -> None: balance, equity = self.balance_equity() self.pm.snapshot(self.account_number, self.server, balance, equity) date_str, pnl = self.today_pnl(balance) self.pm.daily_pnl(self.account_number, self.server, date_str, pnl) print(f"[ok] {datetime.now(JST):%Y-%m-%d %H:%M:%S} JST balance={balance:.2f} equity={equity:.2f} 本日損益={pnl:+.2f}") def register(self) -> None: balance, equity = self.balance_equity() res = self.pm.register(self.account_number, self.server, self.currency, balance, equity, is_demo=self.env_name != "live") linked = res.get("linked") or {} msg = f"紐づけ済み: {linked.get('firm_name')} {linked.get('plan_name')}" if res.get("auto_linked") \ else "未紐づけ（収支ページから手動で紐づけ可能）" print(f"[ok] 口座登録 account={self.account_number} server={self.server} currency={self.currency} / {msg}") def main() -> None: once = "--once" in sys.argv interval = int(_env("PROPMEMO_INTERVAL", "60") or 60) tracker = Tracker() tracker.connect() tracker.register() if once: tracker.tick() return print(f"常駐ループ開始（{interval}秒間隔）。停止: Ctrl+C") failures = 0 while True: try: tracker.tick() failures = 0 except KeyboardInterrupt: print("停止しました。") return except Exception as e: # noqa: BLE001 failures += 1 print(f"[error] 送信失敗({failures}回連続): {e}") if failures >= RECONNECT_AFTER_FAILURES: try: tracker.connect() failures = 0 print("[ok] TradeLockerへ再接続しました") except Exception as e2: # noqa: BLE001 print(f"[error] 再接続失敗: {e2}") try: time.sleep(interval) except KeyboardInterrupt: print("停止しました。") return if __name__ == "__main__": main()