""" client.py — Capital.com REST API client. ========================================= Handles all HTTP communication with Capital.com. Session management, authentication, and all API calls live here. API NOTES: ---------- - Authentication: POST /session with API key + credentials Returns CST and X-SECURITY-TOKEN headers, valid for 10 minutes. - All subsequent requests need CST + X-SECURITY-TOKEN headers. - Rate limit: 1 request per 0.1s for order operations. - Working orders use field names: workingOrderData.orderLevel, orderSize, dealId - Positions use field names: position.level, position.size, position.profitLevel - Account balance: accounts[].balance.balance (equity), .deposit (funds), .available - Price snapshot: snapshot.bid, snapshot.offer (mid = average of both) FIELD NAME REFERENCE (verified against live API): -------------------------------------------------- Working orders response: o["workingOrderData"]["dealId"] — order ID o["workingOrderData"]["orderLevel"] — entry price o["workingOrderData"]["orderSize"] — position size in shares o["workingOrderData"]["profitDistance"] — TP distance in points Positions response: p["position"]["dealId"] — position ID p["position"]["level"] — entry price p["position"]["size"] — position size in shares p["position"]["profitLevel"] — take profit price (None if not set) p["position"]["upl"] — unrealised P&L in USD Accounts response: acc["balance"]["balance"] — current equity (GBP) acc["balance"]["deposit"] — deposited funds (GBP) acc["balance"]["available"] — available to trade (GBP) """ import time import logging from typing import Optional import requests import config log = logging.getLogger("maxbot.client") class CapitalClient: """ Wraps all Capital.com API calls. Handles session creation, token refresh, and error handling. """ def __init__(self): self.base = config.DEMO_URL if config.USE_DEMO else config.BASE_URL self.cst = None self.security_token = None self.session_started = None # ───────────────────────────────────────── # SESSION MANAGEMENT # ───────────────────────────────────────── def create_session(self): """Authenticate and store session tokens.""" url = f"{self.base}/api/v1/session" headers = { "Content-Type": "application/json", "X-CAP-API-KEY": config.API_KEY, } body = { "identifier": config.IDENTIFIER, "password": config.PASSWORD, } resp = requests.post(url, headers=headers, json=body, timeout=15) resp.raise_for_status() self.cst = resp.headers.get("CST") self.security_token = resp.headers.get("X-SECURITY-TOKEN") self.session_started = time.time() log.info("Session created successfully.") def refresh_session_if_needed(self): """Refresh session if tokens are approaching expiry (10 min lifetime).""" if self.session_started is None: self.create_session() return age = time.time() - self.session_started if age >= config.SESSION_REFRESH_SECS: log.info("Refreshing session tokens...") self.create_session() # ───────────────────────────────────────── # HTTP HELPERS # ───────────────────────────────────────── def _headers(self) -> dict: h = { "Content-Type": "application/json", "X-CAP-API-KEY": config.API_KEY, } if self.cst: h["CST"] = self.cst if self.security_token: h["X-SECURITY-TOKEN"] = self.security_token return h def _request(self, method: str, path: str, body: dict = None) -> dict: """Make an authenticated API request. Auto-retries once on 401.""" url = f"{self.base}/api/v1{path}" try: resp = requests.request( method, url, headers=self._headers(), json=body, timeout=15 ) if resp.status_code == 401: log.warning("401 received — refreshing session and retrying...") self.create_session() resp = requests.request( method, url, headers=self._headers(), json=body, timeout=15 ) resp.raise_for_status() return resp.json() if resp.text else {} except requests.RequestException as e: log.error(f"API error: {method} {path} — {e}") raise # ───────────────────────────────────────── # DATA FETCHERS # ───────────────────────────────────────── def get_positions(self) -> list: """Return list of all open positions.""" return self._request("GET", "/positions").get("positions", []) def get_working_orders(self) -> list: """Return list of all pending working orders.""" return self._request("GET", "/workingorders").get("workingOrders", []) def get_account_details(self) -> dict: """Return account details including balance.""" return self._request("GET", "/accounts") def get_price(self, epic: str) -> dict: """Return market snapshot for an epic (bid, offer, status etc).""" return self._request("GET", f"/markets/{epic}") def get_gbpusd(self) -> float: """Fetch live GBP/USD mid rate. Falls back to config value on failure.""" try: data = self._request("GET", "/markets/GBPUSD") snap = data.get("snapshot", {}) return (float(snap["bid"]) + float(snap["offer"])) / 2.0 except Exception: log.warning(f"GBP/USD fetch failed — using fallback {config.GBPUSD_FALLBACK}") return config.GBPUSD_FALLBACK # ───────────────────────────────────────── # ORDER OPERATIONS # ───────────────────────────────────────── def place_limit_order(self, size: float, limit_price: float, take_profit: float) -> dict: """ Place a limit buy order for TSLA. Order fills automatically when price reaches limit_price. take_profit is the absolute price level (not distance). """ body = { "epic": config.EPIC, "direction": "BUY", "size": size, "level": round(limit_price, 2), "type": "LIMIT", "guaranteedStop": False, "profitLevel": round(take_profit, 2), } return self._request("POST", "/workingorders", body) def cancel_order(self, deal_id: str) -> dict: """Cancel a pending working order by deal ID.""" return self._request("DELETE", f"/workingorders/{deal_id}") def update_position_tp(self, deal_id: str, take_profit: Optional[float]) -> dict: """ Update the take profit on an open position. Pass take_profit=None to remove the TP (manual close only). When removing TP, omits the key entirely rather than sending null, as Capital.com may reject a null profitLevel value. """ if take_profit is not None: body = {"profitLevel": round(take_profit, 2)} else: body = {"profitLevel": None} # Capital.com accepts null to clear TP return self._request("PUT", f"/positions/{deal_id}", body)