Krikorios
48e60d015f
- Add advanced metrics dashboard with trade analytics - Add new trading components (EntryTypeAnalysis, MultiDayPositionTracker, NewsEventTracker, etc.) - Add strategy mode selector and trend confirmation - Add risk automation panel and slippage correlation analysis - Add daily trading plan enhancements with modal components - Add custom hooks (useApi, useLocalStorage, useAdvancedTradeMetrics) - Add broker service integration and trading API - Add test setup and vitest configuration - Include parquet data files for live market data - Add comprehensive documentation in docs/ folder
474 lines
17 KiB
Python
474 lines
17 KiB
Python
from __future__ import annotations
|
|
|
|
import abc
|
|
import asyncio
|
|
import uuid
|
|
from dataclasses import asdict, dataclass
|
|
from datetime import datetime, timezone
|
|
from typing import Any, Dict, List, Optional
|
|
|
|
import httpx
|
|
|
|
from app.config import settings
|
|
|
|
try: # Optional dependency for MetaTrader5
|
|
import MetaTrader5 # type: ignore
|
|
except ImportError: # pragma: no cover - optional
|
|
MetaTrader5 = None # type: ignore
|
|
|
|
|
|
class BrokerError(RuntimeError):
|
|
"""Raised when bridge operations fail."""
|
|
|
|
|
|
@dataclass
|
|
class BrokerProvider:
|
|
id: str
|
|
name: str
|
|
description: str
|
|
docs_url: str
|
|
latency_ms: int
|
|
features: Dict[str, bool]
|
|
supports_demo: bool = True
|
|
|
|
|
|
BROKER_PROVIDERS: List[BrokerProvider] = [
|
|
BrokerProvider(
|
|
id="mt5",
|
|
name="MetaTrader 5",
|
|
description="Direct bridge to a locally running MetaTrader 5 terminal.",
|
|
docs_url="https://www.metatrader5.com/en/terminal/help",
|
|
latency_ms=180,
|
|
features={
|
|
"trailingStops": True,
|
|
"partialCloses": True,
|
|
"hedging": True,
|
|
"streaming": True,
|
|
},
|
|
),
|
|
BrokerProvider(
|
|
id="oanda",
|
|
name="OANDA v20",
|
|
description="REST trading for FX/CFD (practice or live)",
|
|
docs_url="https://developer.oanda.com/rest-live-v20/",
|
|
latency_ms=230,
|
|
features={
|
|
"trailingStops": True,
|
|
"partialCloses": True,
|
|
"hedging": False,
|
|
"streaming": False,
|
|
},
|
|
),
|
|
BrokerProvider(
|
|
id="alpaca",
|
|
name="Alpaca Trading",
|
|
description="Equities/crypto order routing (paper or live)",
|
|
docs_url="https://alpaca.markets/docs/api-references/trading-api/",
|
|
latency_ms=120,
|
|
features={
|
|
"trailingStops": False,
|
|
"partialCloses": True,
|
|
"hedging": False,
|
|
"streaming": True,
|
|
},
|
|
),
|
|
]
|
|
|
|
PROVIDER_LOOKUP = {provider.id: provider for provider in BROKER_PROVIDERS}
|
|
|
|
|
|
def _iso_now() -> str:
|
|
return datetime.now(timezone.utc).isoformat()
|
|
|
|
|
|
def _demo_state(balance: Optional[float] = None) -> Dict[str, Any]:
|
|
return {
|
|
"mode": "demo",
|
|
"token": f"demo-{uuid.uuid4()}",
|
|
"balance": balance if balance is not None else settings.BROKER_SIM_BALANCE,
|
|
"positions": [],
|
|
}
|
|
|
|
|
|
class BaseConnector(abc.ABC):
|
|
provider_id: str
|
|
|
|
@abc.abstractmethod
|
|
async def connect(self, credentials: Dict[str, Any]) -> Dict[str, Any]:
|
|
...
|
|
|
|
@abc.abstractmethod
|
|
async def disconnect(self, state: Dict[str, Any]) -> None:
|
|
...
|
|
|
|
@abc.abstractmethod
|
|
async def place_order(self, state: Dict[str, Any], order: Dict[str, Any]) -> Dict[str, Any]:
|
|
...
|
|
|
|
@abc.abstractmethod
|
|
async def sync_positions(self, state: Dict[str, Any]) -> Dict[str, Any]:
|
|
...
|
|
|
|
|
|
class MetaTraderConnector(BaseConnector):
|
|
provider_id = "mt5"
|
|
|
|
def __init__(self) -> None:
|
|
self._lock = asyncio.Lock()
|
|
|
|
def _client(self):
|
|
if MetaTrader5 is None:
|
|
raise BrokerError("MetaTrader5 python package is not installed")
|
|
return MetaTrader5
|
|
|
|
async def connect(self, credentials: Dict[str, Any]) -> Dict[str, Any]:
|
|
if credentials.get("demo", True):
|
|
return _demo_state()
|
|
|
|
mt5 = self._client()
|
|
login = int(credentials["account_id"])
|
|
password = credentials["api_key"]
|
|
server = credentials.get("server") or settings.MT5_SERVER
|
|
|
|
async with self._lock:
|
|
def _login():
|
|
if not mt5.initialize():
|
|
raise BrokerError(f"MetaTrader5 initialize failed: {mt5.last_error()}")
|
|
if not mt5.login(login=login, password=password, server=server):
|
|
raise BrokerError(f"MetaTrader5 login failed: {mt5.last_error()}")
|
|
info = mt5.account_info()
|
|
balance = float(info.balance) if info else None
|
|
return {
|
|
"mode": "live",
|
|
"balance": balance,
|
|
"token": f"mt5-{uuid.uuid4()}",
|
|
}
|
|
|
|
return await asyncio.to_thread(_login)
|
|
|
|
async def disconnect(self, state: Dict[str, Any]) -> None:
|
|
if state.get("mode") == "demo":
|
|
return
|
|
mt5 = self._client()
|
|
|
|
async with self._lock:
|
|
def _shutdown():
|
|
mt5.shutdown()
|
|
|
|
await asyncio.to_thread(_shutdown)
|
|
|
|
async def place_order(self, state: Dict[str, Any], order: Dict[str, Any]) -> Dict[str, Any]:
|
|
if state.get("mode") == "demo":
|
|
return {
|
|
"remote_id": f"demo-{order['action']}-{uuid.uuid4().hex[:6]}",
|
|
"filled": True,
|
|
}
|
|
|
|
mt5 = self._client()
|
|
|
|
def _send():
|
|
request = {
|
|
"action": mt5.TRADE_ACTION_DEAL,
|
|
"symbol": order["symbol"],
|
|
"type": mt5.ORDER_TYPE_BUY if order["action"] == "BUY" else mt5.ORDER_TYPE_SELL,
|
|
"volume": float(order["quantity"]),
|
|
"price": float(order["price"]),
|
|
"type_filling": mt5.ORDER_FILLING_RETURN,
|
|
"sl": order.get("stopLoss"),
|
|
"tp": order.get("takeProfit"),
|
|
}
|
|
result = mt5.order_send(request)
|
|
if result is None or result.retcode != mt5.TRADE_RETCODE_DONE:
|
|
raise BrokerError(f"MetaTrader5 order failed: {mt5.last_error()}")
|
|
return {
|
|
"remote_id": str(result.order),
|
|
"filled": True,
|
|
}
|
|
|
|
return await asyncio.to_thread(_send)
|
|
|
|
async def sync_positions(self, state: Dict[str, Any]) -> Dict[str, Any]:
|
|
if state.get("mode") == "demo":
|
|
return {"positions": state.setdefault("positions", []), "balance": state.get("balance")}
|
|
|
|
mt5 = self._client()
|
|
|
|
def _fetch():
|
|
info = mt5.account_info()
|
|
balance = float(info.balance) if info else None
|
|
rows = mt5.positions_get()
|
|
positions: List[Dict[str, Any]] = []
|
|
if rows:
|
|
for row in rows:
|
|
positions.append(
|
|
{
|
|
"symbol": row.symbol,
|
|
"quantity": float(row.volume),
|
|
"avgPrice": float(row.price_open),
|
|
"lastPrice": float(row.price_current),
|
|
"pnl": float(row.profit),
|
|
"ticket": int(row.ticket),
|
|
}
|
|
)
|
|
return {"positions": positions, "balance": balance}
|
|
|
|
return await asyncio.to_thread(_fetch)
|
|
|
|
|
|
class OandaConnector(BaseConnector):
|
|
provider_id = "oanda"
|
|
|
|
async def connect(self, credentials: Dict[str, Any]) -> Dict[str, Any]:
|
|
if credentials.get("demo", True) or not credentials.get("api_key"):
|
|
return _demo_state()
|
|
|
|
account_id = credentials["account_id"]
|
|
headers = {
|
|
"Authorization": f"Bearer {credentials['api_key']}",
|
|
"Content-Type": "application/json",
|
|
}
|
|
base_url = settings.OANDA_BASE_URL.rstrip("/")
|
|
async with httpx.AsyncClient(base_url=base_url, timeout=settings.BROKER_HTTP_TIMEOUT) as client:
|
|
resp = await client.get(f"/v3/accounts/{account_id}", headers=headers)
|
|
resp.raise_for_status()
|
|
data = resp.json().get("account", {})
|
|
balance = float(data.get("balance", 0))
|
|
return {
|
|
"mode": "live",
|
|
"headers": headers,
|
|
"account_id": account_id,
|
|
"base_url": base_url,
|
|
"balance": balance,
|
|
"token": f"oanda-{uuid.uuid4()}",
|
|
}
|
|
|
|
async def disconnect(self, state: Dict[str, Any]) -> None:
|
|
return None
|
|
|
|
async def place_order(self, state: Dict[str, Any], order: Dict[str, Any]) -> Dict[str, Any]:
|
|
if state.get("mode") == "demo":
|
|
return {
|
|
"remote_id": f"demo-{order['action']}-{uuid.uuid4().hex[:6]}",
|
|
"filled": True,
|
|
}
|
|
|
|
payload = {
|
|
"order": {
|
|
"instrument": order["symbol"],
|
|
"units": str(order["quantity"] if order["action"] == "BUY" else -order["quantity"]),
|
|
"type": order.get("type", "MARKET"),
|
|
"timeInForce": "FOK",
|
|
"positionFill": "DEFAULT",
|
|
}
|
|
}
|
|
if order.get("stopLoss"):
|
|
payload["order"]["stopLossOnFill"] = {"price": str(order["stopLoss"])}
|
|
if order.get("takeProfit"):
|
|
payload["order"]["takeProfitOnFill"] = {"price": str(order["takeProfit"])}
|
|
|
|
async with httpx.AsyncClient(base_url=state["base_url"], timeout=settings.BROKER_HTTP_TIMEOUT) as client:
|
|
resp = await client.post(
|
|
f"/v3/accounts/{state['account_id']}/orders",
|
|
headers=state["headers"],
|
|
json=payload,
|
|
)
|
|
resp.raise_for_status()
|
|
data = resp.json()
|
|
return {
|
|
"remote_id": data.get("orderFillTransaction", {}).get("orderID") or uuid.uuid4().hex,
|
|
"filled": True,
|
|
}
|
|
|
|
async def sync_positions(self, state: Dict[str, Any]) -> Dict[str, Any]:
|
|
if state.get("mode") == "demo":
|
|
return {"positions": state.setdefault("positions", []), "balance": state.get("balance")}
|
|
|
|
async with httpx.AsyncClient(base_url=state["base_url"], timeout=settings.BROKER_HTTP_TIMEOUT) as client:
|
|
resp = await client.get(
|
|
f"/v3/accounts/{state['account_id']}/openPositions",
|
|
headers=state["headers"],
|
|
)
|
|
resp.raise_for_status()
|
|
payload = resp.json()
|
|
|
|
positions: List[Dict[str, Any]] = []
|
|
for item in payload.get("positions", []):
|
|
net = float(item.get("net", {}).get("units", 0))
|
|
if net == 0:
|
|
continue
|
|
avg_price = float(item.get("net", {}).get("averagePrice", 0))
|
|
positions.append(
|
|
{
|
|
"symbol": item.get("instrument"),
|
|
"quantity": abs(net),
|
|
"avgPrice": avg_price,
|
|
"lastPrice": None,
|
|
"pnl": None,
|
|
}
|
|
)
|
|
|
|
return {"positions": positions, "balance": state.get("balance")}
|
|
|
|
|
|
class AlpacaConnector(BaseConnector):
|
|
provider_id = "alpaca"
|
|
|
|
async def connect(self, credentials: Dict[str, Any]) -> Dict[str, Any]:
|
|
if credentials.get("demo", True) or not credentials.get("api_key"):
|
|
return _demo_state()
|
|
|
|
key_parts = credentials["api_key"].split(":", 1)
|
|
if len(key_parts) != 2:
|
|
raise BrokerError("Provide API_KEY:API_SECRET for Alpaca API key field")
|
|
|
|
headers = {
|
|
"APCA-API-KEY-ID": key_parts[0],
|
|
"APCA-API-SECRET-KEY": key_parts[1],
|
|
"Content-Type": "application/json",
|
|
}
|
|
base_url = settings.ALPACA_BASE_URL.rstrip("/")
|
|
async with httpx.AsyncClient(base_url=base_url, timeout=settings.BROKER_HTTP_TIMEOUT) as client:
|
|
resp = await client.get("/account", headers=headers)
|
|
resp.raise_for_status()
|
|
data = resp.json()
|
|
return {
|
|
"mode": "live",
|
|
"headers": headers,
|
|
"base_url": base_url,
|
|
"account_id": data.get("id") or credentials.get("account_id"),
|
|
"balance": float(data.get("cash", 0)),
|
|
"token": f"alpaca-{uuid.uuid4()}",
|
|
}
|
|
|
|
async def disconnect(self, state: Dict[str, Any]) -> None:
|
|
return None
|
|
|
|
async def place_order(self, state: Dict[str, Any], order: Dict[str, Any]) -> Dict[str, Any]:
|
|
if state.get("mode") == "demo":
|
|
return {
|
|
"remote_id": f"demo-{order['action']}-{uuid.uuid4().hex[:6]}",
|
|
"filled": True,
|
|
}
|
|
|
|
payload = {
|
|
"symbol": order["symbol"],
|
|
"qty": order["quantity"],
|
|
"side": "buy" if order["action"] == "BUY" else "sell",
|
|
"type": order.get("type", "market").lower(),
|
|
"time_in_force": "day",
|
|
}
|
|
if order.get("stopLoss") or order.get("takeProfit"):
|
|
payload["order_class"] = "oto"
|
|
payload["take_profit"] = {"limit_price": order.get("takeProfit")}
|
|
payload["stop_loss"] = {"stop_price": order.get("stopLoss")}
|
|
|
|
async with httpx.AsyncClient(base_url=state["base_url"], timeout=settings.BROKER_HTTP_TIMEOUT) as client:
|
|
resp = await client.post("/orders", headers=state["headers"], json=payload)
|
|
resp.raise_for_status()
|
|
data = resp.json()
|
|
return {
|
|
"remote_id": data.get("id", uuid.uuid4().hex),
|
|
"filled": data.get("status") == "filled",
|
|
}
|
|
|
|
async def sync_positions(self, state: Dict[str, Any]) -> Dict[str, Any]:
|
|
if state.get("mode") == "demo":
|
|
return {"positions": state.setdefault("positions", []), "balance": state.get("balance")}
|
|
|
|
async with httpx.AsyncClient(base_url=state["base_url"], timeout=settings.BROKER_HTTP_TIMEOUT) as client:
|
|
resp = await client.get("/positions", headers=state["headers"])
|
|
resp.raise_for_status()
|
|
rows = resp.json()
|
|
positions = [
|
|
{
|
|
"symbol": row.get("symbol"),
|
|
"quantity": float(row.get("qty", 0)),
|
|
"avgPrice": float(row.get("avg_entry_price", 0)),
|
|
"lastPrice": float(row.get("current_price", 0)),
|
|
"pnl": float(row.get("unrealized_pl", 0)),
|
|
}
|
|
for row in rows
|
|
]
|
|
return {"positions": positions, "balance": state.get("balance")}
|
|
|
|
|
|
CONNECTORS: Dict[str, BaseConnector] = {
|
|
"mt5": MetaTraderConnector(),
|
|
"oanda": OandaConnector(),
|
|
"alpaca": AlpacaConnector(),
|
|
}
|
|
|
|
|
|
class BrokerBridgeService:
|
|
def __init__(self) -> None:
|
|
self._session: Optional[Dict[str, Any]] = None
|
|
self._lock = asyncio.Lock()
|
|
|
|
def list_providers(self) -> List[Dict[str, Any]]:
|
|
return [asdict(provider) for provider in BROKER_PROVIDERS]
|
|
|
|
def get_session(self) -> Optional[Dict[str, Any]]:
|
|
if not self._session:
|
|
return None
|
|
provider = PROVIDER_LOOKUP.get(self._session["provider_id"])
|
|
payload = {**self._session}
|
|
payload["provider"] = asdict(provider) if provider else None
|
|
return payload
|
|
|
|
async def connect(self, provider_id: str, credentials: Dict[str, Any]) -> Dict[str, Any]:
|
|
connector = CONNECTORS.get(provider_id)
|
|
if not connector:
|
|
raise BrokerError("Unsupported broker provider")
|
|
|
|
state = await connector.connect(credentials)
|
|
async with self._lock:
|
|
self._session = {
|
|
"provider_id": provider_id,
|
|
"credentials": credentials,
|
|
"state": state,
|
|
"account_id": credentials.get("account_id"),
|
|
"demo": credentials.get("demo", True),
|
|
"last_heartbeat": _iso_now(),
|
|
"balance": state.get("balance"),
|
|
"positions": state.get("positions", []),
|
|
}
|
|
return self.get_session() # type: ignore[return-value]
|
|
|
|
async def disconnect(self) -> None:
|
|
if not self._session:
|
|
return
|
|
connector = CONNECTORS.get(self._session["provider_id"])
|
|
if connector:
|
|
await connector.disconnect(self._session.get("state", {}))
|
|
async with self._lock:
|
|
self._session = None
|
|
|
|
async def place_order(self, order: Dict[str, Any]) -> Dict[str, Any]:
|
|
if not self._session:
|
|
raise BrokerError("No active broker session")
|
|
connector = CONNECTORS.get(self._session["provider_id"])
|
|
if not connector:
|
|
raise BrokerError("Unsupported broker provider")
|
|
result = await connector.place_order(self._session.get("state", {}), order)
|
|
self._session["last_heartbeat"] = _iso_now()
|
|
return result
|
|
|
|
async def sync_positions(self) -> Dict[str, Any]:
|
|
if not self._session:
|
|
raise BrokerError("No active broker session")
|
|
connector = CONNECTORS.get(self._session["provider_id"])
|
|
if not connector:
|
|
raise BrokerError("Unsupported broker provider")
|
|
snapshot = await connector.sync_positions(self._session.get("state", {}))
|
|
self._session["last_heartbeat"] = _iso_now()
|
|
self._session["positions"] = snapshot.get("positions", [])
|
|
self._session["balance"] = snapshot.get("balance", self._session.get("balance"))
|
|
return {
|
|
"positions": self._session["positions"],
|
|
"balance": self._session.get("balance"),
|
|
"lastHeartbeat": self._session.get("last_heartbeat"),
|
|
}
|
|
|
|
|
|
broker_bridge_service = BrokerBridgeService()
|