"""TJP Cloud Python SDK - standard library only (Python 3.9+). from tjpcloud import TjpCloud tjp = TjpCloud(api_key=os.environ["TJP_API_KEY"]) account = tjp.accounts()[0] tjp.place_order(account["id"], side="buy", symbol="EURUSD", volume=0.1, stop_loss=1.08) Every write sends an Idempotency-Key (generated unless you pass one). If a write raises TjpCloudError with uncertain=True, the broker may have executed it: check positions, or retry with the SAME idempotency_key - never a new one. """ from __future__ import annotations import hashlib import hmac import json import re import time import uuid from typing import Any, Dict, List, Optional from urllib import error, parse, request __all__ = ["TjpCloud", "TjpCloudError", "verify_webhook"] __version__ = "1.0.0" _KEY = re.compile(r"^tjpc_live_[0-9a-f]{12}_[A-Za-z0-9_-]{32}$") class TjpCloudError(Exception): def __init__(self, status: int, code: str, message: str, uncertain: bool = False, retcode: Optional[int] = None, retry_after: Optional[int] = None): super().__init__(message) self.status, self.code, self.uncertain, self.retcode, self.retry_after = status, code, uncertain, retcode, retry_after class TjpCloud: def __init__(self, api_key: str, base_url: str = "https://api.tradejournalpro.net", timeout: float = 45.0): if not _KEY.match(api_key): raise ValueError("A TJP Cloud API key looks like tjpc_live_<12 hex>_<32 chars>.") self.api_key, self.base_url, self.timeout = api_key, base_url.rstrip("/"), timeout # --- Accounts and data ------------------------------------------------- def accounts(self) -> List[Dict[str, Any]]: return self._get("/v1/accounts")["accounts"] def account(self, account_id: str) -> Dict[str, Any]: return self._get(f"/v1/accounts/{_q(account_id)}")["account"] def positions(self, account_id: str) -> List[Dict[str, Any]]: return self._get(f"/v1/accounts/{_q(account_id)}/positions")["positions"] def orders(self, account_id: str) -> List[Dict[str, Any]]: return self._get(f"/v1/accounts/{_q(account_id)}/orders")["orders"] def symbols(self, account_id: str) -> List[str]: return self._get(f"/v1/accounts/{_q(account_id)}/symbols")["symbols"] def symbol(self, account_id: str, symbol: str) -> Dict[str, Any]: return self._get(f"/v1/accounts/{_q(account_id)}/symbols/{_q(symbol)}")["symbol"] def quote(self, account_id: str, symbol: str) -> Dict[str, Any]: return self._get(f"/v1/accounts/{_q(account_id)}/quotes/{_q(symbol)}")["quote"] def candles(self, account_id: str, symbol: str, timeframe: str = "H1", start: Optional[str] = None, end: Optional[str] = None, count: Optional[int] = None) -> List[Dict[str, Any]]: query = _query(symbol=symbol, timeframe=timeframe, **{"from": start, "to": end}, count=count) return self._get(f"/v1/accounts/{_q(account_id)}/candles?{query}")["candles"] def deals(self, account_id: str, start: Optional[str] = None, end: Optional[str] = None) -> List[Dict[str, Any]]: return self._get(f"/v1/accounts/{_q(account_id)}/deals?{_query(**{'from': start, 'to': end})}")["deals"] def history_orders(self, account_id: str, start: Optional[str] = None, end: Optional[str] = None) -> List[Dict[str, Any]]: return self._get(f"/v1/accounts/{_q(account_id)}/history-orders?{_query(**{'from': start, 'to': end})}")["orders"] # --- Trading (idempotent) --------------------------------------------- def place_order(self, account_id: str, side: str, symbol: str, volume: float, type: str = "market", price: Optional[float] = None, stop_loss: Optional[float] = None, take_profit: Optional[float] = None, expires_at: Optional[str] = None, client_order_id: Optional[str] = None, idempotency_key: Optional[str] = None) -> Dict[str, Any]: body = _drop_none(side=side, type=type, symbol=symbol, volume=volume, price=price, stopLoss=stop_loss, takeProfit=take_profit, expiresAt=expires_at, clientOrderId=client_order_id) return self._write("POST", f"/v1/accounts/{_q(account_id)}/orders", body, idempotency_key)["result"] def modify_position(self, account_id: str, ticket: str, idempotency_key: Optional[str] = None, **change: Any) -> Dict[str, Any]: """Pass stop_loss and/or take_profit; None removes a level, leaving one out keeps it.""" body = {_camel(k): v for k, v in change.items() if k in ("stop_loss", "take_profit")} return self._write("PATCH", f"/v1/accounts/{_q(account_id)}/positions/{_q(ticket)}", body, idempotency_key)["result"] def close_position(self, account_id: str, ticket: str, volume: Optional[float] = None, idempotency_key: Optional[str] = None) -> Dict[str, Any]: body = None if volume is None else {"volume": volume} return self._write("DELETE", f"/v1/accounts/{_q(account_id)}/positions/{_q(ticket)}", body, idempotency_key)["result"] def close_all(self, account_id: str, symbol: Optional[str] = None, side: Optional[str] = None, idempotency_key: Optional[str] = None) -> List[Dict[str, Any]]: query = _query(symbol=symbol, side=side) return self._write("DELETE", f"/v1/accounts/{_q(account_id)}/positions" + (f"?{query}" if query else ""), None, idempotency_key)["results"] def modify_order(self, account_id: str, ticket: str, idempotency_key: Optional[str] = None, **change: Any) -> Dict[str, Any]: """Pass price, stop_loss, take_profit and/or expires_at (None = good till cancelled).""" body = {_camel(k): v for k, v in change.items() if k in ("price", "stop_loss", "take_profit", "expires_at")} return self._write("PATCH", f"/v1/accounts/{_q(account_id)}/orders/{_q(ticket)}", body, idempotency_key)["result"] def cancel_order(self, account_id: str, ticket: str, idempotency_key: Optional[str] = None) -> Dict[str, Any]: return self._write("DELETE", f"/v1/accounts/{_q(account_id)}/orders/{_q(ticket)}", None, idempotency_key)["result"] def set_rules(self, account_id: str, ticket: str, breakeven: Optional[Dict[str, Any]] = None, trailing: Optional[Dict[str, Any]] = None, take_profits: Optional[List[Dict[str, Any]]] = None, idempotency_key: Optional[str] = None) -> Dict[str, Any]: """Breakeven, trailing stop and partial take-profits run by the terminal (MT5). Points are the symbol's point size. breakeven={"triggerPoints": 150, "offsetPoints": 10}, trailing={"startPoints": 200, "distancePoints": 120, "stepPoints": 20}, take_profits=[{"price": 1.09, "volume": 0.05}, {"price": 1.1}] """ body = _drop_none(breakeven=breakeven, trailing=trailing, takeProfits=take_profits) return self._write("PUT", f"/v1/accounts/{_q(account_id)}/positions/{_q(ticket)}/rules", body, idempotency_key)["result"] def rules(self, account_id: str) -> List[Dict[str, Any]]: return self._get(f"/v1/accounts/{_q(account_id)}/rules")["rules"] def clear_rules(self, account_id: str, ticket: str, idempotency_key: Optional[str] = None) -> Dict[str, Any]: return self._write("DELETE", f"/v1/accounts/{_q(account_id)}/positions/{_q(ticket)}/rules", None, idempotency_key)["result"] def bulk_orders(self, account_ids: List[str], order: Dict[str, Any], idempotency_key: Optional[str] = None) -> List[Dict[str, Any]]: return self._write("POST", "/v1/bulk/orders", {"accountIds": account_ids, "order": order}, idempotency_key)["results"] # --- Plumbing ---------------------------------------------------------- def _get(self, path: str) -> Dict[str, Any]: return self._request("GET", path) def _write(self, method: str, path: str, body: Optional[Dict[str, Any]], key: Optional[str]) -> Dict[str, Any]: return self._request(method, path, body, key or f"sdk-{uuid.uuid4()}") def _request(self, method: str, path: str, body: Optional[Dict[str, Any]] = None, key: Optional[str] = None) -> Dict[str, Any]: headers = {"Authorization": f"Bearer {self.api_key}", "Accept": "application/json", "User-Agent": f"tjpcloud-python/{__version__}"} data = None if body is not None: data = json.dumps(body).encode() headers["Content-Type"] = "application/json" if key: headers["Idempotency-Key"] = key req = request.Request(self.base_url + path, data=data, method=method, headers=headers) write = method != "GET" try: with request.urlopen(req, timeout=self.timeout) as response: return json.loads(response.read() or b"{}") except error.HTTPError as failure: try: payload = json.loads(failure.read() or b"{}") except ValueError: payload = {} info = payload.get("error") or {} retry_after = failure.headers.get("Retry-After") raise TjpCloudError(failure.code, info.get("code", "HTTP_ERROR"), info.get("message", f"HTTP {failure.code}"), payload.get("uncertain") is True, info.get("retcode"), int(retry_after) if retry_after and retry_after.isdigit() else None) from None except OSError as failure: # URLError, timeouts, resets and aborted connections prefix = "No answer for this write: it may have been executed. Retry with the same idempotency_key. " if write else "" raise TjpCloudError(0, "NETWORK", prefix + str(failure), uncertain=write) from None def verify_webhook(secret: str, raw_body: bytes, signature_header: str, tolerance_sec: int = 300, now: Optional[float] = None) -> bool: """Checks X-TJP-Signature (t=,v1=) against the raw request body.""" parts = dict(part.split("=", 1) for part in signature_header.split(",") if "=" in part) try: timestamp = int(parts["t"]) except (KeyError, ValueError): return False if abs((now if now is not None else time.time()) - timestamp) > tolerance_sec: return False expected = hmac.new(secret.encode(), f"{parts['t']}.".encode() + raw_body, hashlib.sha256).hexdigest() return hmac.compare_digest(expected, parts.get("v1", "")) def _q(value: str) -> str: return parse.quote(str(value), safe="") def _query(**values: Any) -> str: return parse.urlencode({k: v for k, v in values.items() if v is not None}) def _drop_none(**values: Any) -> Dict[str, Any]: return {k: v for k, v in values.items() if v is not None} def _camel(name: str) -> str: head, *rest = name.split("_") return head + "".join(word.title() for word in rest)