Add watch_ticker and watch_ohlcv (polling-based)
This commit is contained in:
+166
-210
@@ -21,6 +21,7 @@ Architecture:
|
||||
MetaTrader 5 client
|
||||
"""
|
||||
|
||||
import asyncio
|
||||
import json
|
||||
import time
|
||||
from datetime import datetime, timezone
|
||||
@@ -87,28 +88,26 @@ MT5_ORDER_TYPE = {
|
||||
}
|
||||
|
||||
# MT5 retcode -> CCXT-friendly status string
|
||||
# Source: MQL5 docs, ENUM_TRADE_REQUEST_RETCODE
|
||||
MT5_RETCODE_STATUS = {
|
||||
10009: "open", # TRADE_RETCODE_DONE - request executed
|
||||
10010: "filled", # TRADE_RETCODE_PLACED - pending order placed
|
||||
10011: "canceled", # TRADE_RETCODE_DONE_PARTIAL
|
||||
10012: "canceled", # TRADE_RETCODE_ERROR
|
||||
10013: "rejected", # TRADE_RETCODE_TIMEOUT
|
||||
10014: "rejected", # TRADE_RETCODE_INVALID
|
||||
10015: "rejected", # TRADE_RETCODE_INVALID_VOLUME
|
||||
10016: "rejected", # TRADE_RETCODE_INVALID_PRICE
|
||||
10017: "rejected", # TRADE_RETCODE_BUSY
|
||||
10018: "rejected", # TRADE_RETCODE_NO_CHANGES
|
||||
10019: "rejected", # TRADE_RETCODE_LOCKED
|
||||
10020: "rejected", # TRADE_RETCODE_FROZEN
|
||||
10021: "rejected", # TRADE_RETCODE_INVALID_FILL
|
||||
10022: "rejected", # TRADE_RETCODE_CONNECTION
|
||||
10023: "rejected", # TRADE_RETCODE_ONLY_REAL
|
||||
10024: "rejected", # TRADE_RETCODE_LIMIT_ORDERS
|
||||
10025: "rejected", # TRADE_RETCODE_LIMIT_VOLUME
|
||||
10009: "open",
|
||||
10010: "filled",
|
||||
10011: "canceled",
|
||||
10012: "canceled",
|
||||
10013: "rejected",
|
||||
10014: "rejected",
|
||||
10015: "rejected",
|
||||
10016: "rejected",
|
||||
10017: "rejected",
|
||||
10018: "rejected",
|
||||
10019: "rejected",
|
||||
10020: "rejected",
|
||||
10021: "rejected",
|
||||
10022: "rejected",
|
||||
10023: "rejected",
|
||||
10024: "rejected",
|
||||
10025: "rejected",
|
||||
}
|
||||
|
||||
# MT5 retcode -> True if it indicates insufficient funds
|
||||
MT5_RETCODE_INSUFFICIENT_FUNDS = {10019, 10020, 10021, 10022, 10025}
|
||||
|
||||
|
||||
@@ -124,9 +123,7 @@ class mt5bridge(ccxt.Exchange):
|
||||
secret: not used (placeholder for CCXT compatibility)
|
||||
host: base URL of MT5Bridge (default: http://localhost:8080)
|
||||
timeout: request timeout in ms (default: 30000)
|
||||
symbols: optional dict mapping CCXT symbols to MT5 symbols, e.g.
|
||||
{"XAU/USD": "XAUUSDc", "EUR/USD": "EURUSDc"}
|
||||
If not provided, the wrapper strips "/" from CCXT symbols.
|
||||
symbols: optional dict mapping CCXT symbols to MT5 symbols
|
||||
default_volume_step: default lot rounding (default 0.01)
|
||||
|
||||
Example:
|
||||
@@ -143,7 +140,7 @@ class mt5bridge(ccxt.Exchange):
|
||||
id = "mt5bridge"
|
||||
name = "MT5 Bridge"
|
||||
countries = ["*"]
|
||||
version = "0.1.0"
|
||||
version = "0.2.0"
|
||||
rateLimit = 100
|
||||
certified = False
|
||||
pro = False
|
||||
@@ -185,6 +182,16 @@ class mt5bridge(ccxt.Exchange):
|
||||
"setMarginMode": False,
|
||||
"transfer": False,
|
||||
"withdraw": False,
|
||||
# ── Watching (polling-based, see watch_* methods) ──
|
||||
"watchBalance": False,
|
||||
"watchMyTrades": False,
|
||||
"watchOHLCV": True, # polling-based
|
||||
"watchOrderBook": False,
|
||||
"watchOrders": False,
|
||||
"watchPositions": False,
|
||||
"watchTicker": True, # polling-based
|
||||
"watchTickers": False,
|
||||
"watchTrades": False,
|
||||
}
|
||||
|
||||
timeframes = {
|
||||
@@ -197,6 +204,18 @@ class mt5bridge(ccxt.Exchange):
|
||||
"1d": "TIMEFRAME_D1",
|
||||
}
|
||||
|
||||
# Polling intervals per timeframe (used by watch_ohlcv as the minimum
|
||||
# interval to check for a new bar)
|
||||
WATCH_OHLCV_POLL_MS = {
|
||||
"1m": 5000,
|
||||
"5m": 10000,
|
||||
"15m": 15000,
|
||||
"30m": 30000,
|
||||
"1h": 60000,
|
||||
"4h": 120000,
|
||||
"1d": 300000,
|
||||
}
|
||||
|
||||
urls = {
|
||||
"api": "http://localhost:8080",
|
||||
}
|
||||
@@ -239,16 +258,14 @@ class mt5bridge(ccxt.Exchange):
|
||||
def __init__(self, config={}):
|
||||
super().__init__(config)
|
||||
|
||||
# Configuration
|
||||
self.host = (config.get("host") or self.urls["api"]).rstrip("/")
|
||||
self.apiKey = config.get("apiKey") or self.apiKey or ""
|
||||
self.timeout = config.get("timeout") or 30000
|
||||
self.default_volume_step = config.get("default_volume_step", 0.01)
|
||||
self.symbol_map = config.get("symbols", {}) or {}
|
||||
# Reverse map for displaying MT5 symbols as CCXT symbols
|
||||
self._mt5_to_ccxt = {v: k for k, v in self.symbol_map.items()}
|
||||
|
||||
# HTTP session with retry logic
|
||||
# HTTP session with retry
|
||||
self.session = requests.Session()
|
||||
retry = Retry(
|
||||
total=3,
|
||||
@@ -266,47 +283,25 @@ class mt5bridge(ccxt.Exchange):
|
||||
"User-Agent": f"mt5bridge-ccxt/{self.version}",
|
||||
})
|
||||
|
||||
# MQL5 helper
|
||||
self.mql5 = Mql5Bridge(self)
|
||||
|
||||
# ──────── Internal helpers ────────
|
||||
|
||||
def _resolve_mt5_symbol(self, symbol):
|
||||
"""Convert CCXT symbol (XAU/USD) to MT5 symbol (XAUUSDc)."""
|
||||
if not symbol:
|
||||
return symbol
|
||||
if symbol in self.symbol_map:
|
||||
return self.symbol_map[symbol]
|
||||
# Fallback: strip slash
|
||||
return symbol.replace("/", "")
|
||||
|
||||
def _resolve_ccxt_symbol(self, mt5_symbol):
|
||||
"""Convert MT5 symbol back to CCXT format for display."""
|
||||
if not mt5_symbol:
|
||||
return mt5_symbol
|
||||
if mt5_symbol in self._mt5_to_ccxt:
|
||||
return self._mt5_to_ccxt[mt5_symbol]
|
||||
# Best-effort fallback
|
||||
return mt5_symbol
|
||||
|
||||
def _request(self, method, path, params=None, json_body=None):
|
||||
"""Make HTTP request to MT5Bridge with proper error handling.
|
||||
|
||||
Args:
|
||||
method: HTTP method (GET, POST, DELETE)
|
||||
path: API path (without host)
|
||||
params: query parameters
|
||||
json_body: JSON request body for POST
|
||||
|
||||
Returns:
|
||||
dict: response JSON
|
||||
|
||||
Raises:
|
||||
Mt5BridgeAuthError: HTTP 401
|
||||
Mt5BridgeInvalidRequestError: HTTP 400
|
||||
Mt5BridgeNotConnectedError: HTTP 503
|
||||
Mt5BridgeError: other failures
|
||||
"""
|
||||
url = f"{self.host}/{path.lstrip('/')}"
|
||||
timeout_s = self.timeout / 1000.0
|
||||
|
||||
@@ -329,7 +324,6 @@ class mt5bridge(ccxt.Exchange):
|
||||
except requests.exceptions.RequestException as e:
|
||||
raise Mt5BridgeError(f"Request failed: {e}") from e
|
||||
|
||||
# Map HTTP status codes to exceptions
|
||||
if response.status_code == 401:
|
||||
raise Mt5BridgeAuthError("Unauthorized: check your API key")
|
||||
if response.status_code == 503:
|
||||
@@ -356,7 +350,6 @@ class mt5bridge(ccxt.Exchange):
|
||||
) from e
|
||||
|
||||
def _extract_detail(self, response, fallback):
|
||||
"""Extract 'detail' field from a JSON error response."""
|
||||
try:
|
||||
body = response.json()
|
||||
return body.get("detail") or fallback
|
||||
@@ -364,13 +357,6 @@ class mt5bridge(ccxt.Exchange):
|
||||
return fallback or response.text[:200]
|
||||
|
||||
def _parse_datetime(self, dt_str):
|
||||
"""Parse ISO datetime to millisecond timestamp.
|
||||
|
||||
Handles formats:
|
||||
"yyyy-MM-ddTHH:mm:ss"
|
||||
"yyyy-MM-ddTHH:mm:ss.fff000"
|
||||
"yyyy-MM-dd"
|
||||
"""
|
||||
if not dt_str:
|
||||
return None
|
||||
try:
|
||||
@@ -387,11 +373,9 @@ class mt5bridge(ccxt.Exchange):
|
||||
return None
|
||||
|
||||
def _get_order_type_code(self, order_type, side):
|
||||
"""Map CCXT (order_type, side) to MT5 order type code."""
|
||||
key = (order_type, side)
|
||||
if key in MT5_ORDER_TYPE:
|
||||
return MT5_ORDER_TYPE[key]
|
||||
# Fallback: try the more general mapping
|
||||
valid_types = ("market", "limit", "stop", "stop_limit")
|
||||
if order_type not in valid_types:
|
||||
raise InvalidOrder(
|
||||
@@ -400,15 +384,12 @@ class mt5bridge(ccxt.Exchange):
|
||||
)
|
||||
if side not in ("buy", "sell"):
|
||||
raise InvalidOrder(f"Unsupported side '{side}'. Valid: buy, sell")
|
||||
# Should not reach here given valid_types
|
||||
raise InvalidOrder(f"Unsupported combination: {order_type} + {side}")
|
||||
|
||||
def _get_status_from_retcode(self, retcode):
|
||||
"""Map MT5 retcode to CCXT status string."""
|
||||
return MT5_RETCODE_STATUS.get(int(retcode), "rejected")
|
||||
|
||||
def _format_volume(self, amount):
|
||||
"""Round volume to nearest step."""
|
||||
if amount is None:
|
||||
return 0.0
|
||||
try:
|
||||
@@ -416,14 +397,19 @@ class mt5bridge(ccxt.Exchange):
|
||||
except (TypeError, ValueError):
|
||||
return 0.0
|
||||
|
||||
@staticmethod
|
||||
def safe_float(d, key, default=None):
|
||||
try:
|
||||
v = d.get(key)
|
||||
if v is None:
|
||||
return default
|
||||
return float(v)
|
||||
except (TypeError, ValueError):
|
||||
return default
|
||||
|
||||
# ──────── Public API: status & markets ────────
|
||||
|
||||
def fetch_status(self, params={}):
|
||||
"""Check MT5Bridge and MT5 connection status.
|
||||
|
||||
Returns:
|
||||
dict: {"status": "ok"|"maintenance", "updated": ms, "info": ...}
|
||||
"""
|
||||
try:
|
||||
r = self._request("GET", "/health")
|
||||
except Mt5BridgeNotConnectedError:
|
||||
@@ -443,27 +429,14 @@ class mt5bridge(ccxt.Exchange):
|
||||
}
|
||||
|
||||
def fetch_markets(self, params={}):
|
||||
"""Fetch available markets.
|
||||
|
||||
Note: MT5Bridge doesn't expose a "list all symbols" endpoint.
|
||||
Markets are built from the `symbols` config (CCXT -> MT5 mapping).
|
||||
To add a market, populate the `symbols` config dict.
|
||||
|
||||
Returns:
|
||||
list of market dicts (CCXT format)
|
||||
"""
|
||||
if not self.symbol_map:
|
||||
# Try to discover from positions / tickers
|
||||
return []
|
||||
|
||||
markets = []
|
||||
for ccxt_symbol, mt5_symbol in self.symbol_map.items():
|
||||
markets.append(self._build_market(ccxt_symbol, mt5_symbol))
|
||||
return markets
|
||||
|
||||
def _build_market(self, ccxt_symbol, mt5_symbol):
|
||||
"""Build a CCXT market dict from symbol info."""
|
||||
# Try to get symbol info (optional - don't fail if unavailable)
|
||||
info_data = {}
|
||||
try:
|
||||
r = self._request("GET", f"/symbols/{mt5_symbol}")
|
||||
@@ -484,7 +457,7 @@ class mt5bridge(ccxt.Exchange):
|
||||
"baseId": base,
|
||||
"quoteId": quote,
|
||||
"active": True,
|
||||
"type": "swap", # MT5 OTC/CFD
|
||||
"type": "swap",
|
||||
"spot": False,
|
||||
"margin": True,
|
||||
"swap": True,
|
||||
@@ -513,18 +486,9 @@ class mt5bridge(ccxt.Exchange):
|
||||
"info": info_data,
|
||||
}
|
||||
|
||||
# ──────── Public API: market data ────────
|
||||
# ──────── Public API: market data (fetch_*) ────────
|
||||
|
||||
def fetch_ticker(self, symbol, params={}):
|
||||
"""Fetch current ticker for a symbol.
|
||||
|
||||
Args:
|
||||
symbol: CCXT symbol (e.g. "XAU/USD")
|
||||
params: extra params
|
||||
|
||||
Returns:
|
||||
dict: CCXT ticker
|
||||
"""
|
||||
mt5_symbol = self._resolve_mt5_symbol(symbol)
|
||||
r = self._request("GET", f"/symbols/{mt5_symbol}/tick")
|
||||
data = r["data"][0]
|
||||
@@ -558,20 +522,6 @@ class mt5bridge(ccxt.Exchange):
|
||||
}
|
||||
|
||||
def fetch_ohlcv(self, symbol, timeframe="1h", since=None, limit=None, params={}):
|
||||
"""Fetch OHLCV candlestick data.
|
||||
|
||||
Args:
|
||||
symbol: CCXT symbol (e.g. "XAU/USD")
|
||||
timeframe: "1m" | "5m" | "15m" | "30m" | "1h" | "4h" | "1d"
|
||||
since: start timestamp in milliseconds (optional)
|
||||
limit: max number of bars (default 100, max 10000)
|
||||
params: extra params. Can include:
|
||||
- "date_from": explicit start date "yyyy-MM-dd"
|
||||
- "date_to": explicit end date "yyyy-MM-dd"
|
||||
|
||||
Returns:
|
||||
list of [timestamp, open, high, low, close, volume]
|
||||
"""
|
||||
if timeframe not in self.timeframes:
|
||||
raise BadRequest(
|
||||
f"Unsupported timeframe '{timeframe}'. "
|
||||
@@ -581,7 +531,6 @@ class mt5bridge(ccxt.Exchange):
|
||||
mt5_tf = self.timeframes[timeframe]
|
||||
limit = min(limit or 100, 10000)
|
||||
|
||||
# Decide endpoint
|
||||
if "date_from" in params and "date_to" in params:
|
||||
query = {
|
||||
"symbol": mt5_symbol,
|
||||
@@ -623,24 +572,9 @@ class mt5bridge(ccxt.Exchange):
|
||||
])
|
||||
return result
|
||||
|
||||
@staticmethod
|
||||
def safe_float(d, key, default=None):
|
||||
try:
|
||||
v = d.get(key)
|
||||
if v is None:
|
||||
return default
|
||||
return float(v)
|
||||
except (TypeError, ValueError):
|
||||
return default
|
||||
|
||||
# ──────── Public API: account & positions ────────
|
||||
|
||||
def fetch_balance(self, params={}):
|
||||
"""Fetch account balance.
|
||||
|
||||
Returns:
|
||||
dict: {info, timestamp, free, used, total, debt}
|
||||
"""
|
||||
r = self._request("GET", "/account")
|
||||
acc = r["data"][0]
|
||||
currency = acc.get("currency") or "USD"
|
||||
@@ -657,15 +591,6 @@ class mt5bridge(ccxt.Exchange):
|
||||
}
|
||||
|
||||
def fetch_positions(self, symbols=None, params={}):
|
||||
"""Fetch open positions.
|
||||
|
||||
Args:
|
||||
symbols: optional list of CCXT symbols to filter
|
||||
params: extra params
|
||||
|
||||
Returns:
|
||||
list of CCXT position dicts
|
||||
"""
|
||||
r = self._request("GET", "/positions")
|
||||
positions = r.get("data", [])
|
||||
|
||||
@@ -718,30 +643,14 @@ class mt5bridge(ccxt.Exchange):
|
||||
return result
|
||||
|
||||
def fetch_position(self, symbol, params={}):
|
||||
"""Fetch a single position by symbol.
|
||||
|
||||
Note: MT5 can have multiple positions per symbol; this returns the
|
||||
most recently opened one (or None if no position exists).
|
||||
"""
|
||||
positions = self.fetch_positions([symbol], params)
|
||||
if not positions:
|
||||
return None
|
||||
# Return the one with highest ticket (most recent)
|
||||
return max(positions, key=lambda p: int(p["info"].get("ticket", 0)))
|
||||
|
||||
# ──────── Public API: orders & trades ────────
|
||||
|
||||
def fetch_open_orders(self, symbol=None, since=None, limit=None, params={}):
|
||||
"""Fetch open (pending) orders.
|
||||
|
||||
Args:
|
||||
symbol: optional CCXT symbol to filter
|
||||
since: not used by MT5Bridge
|
||||
limit: not used
|
||||
|
||||
Returns:
|
||||
list of order dicts
|
||||
"""
|
||||
mt5_symbol = self._resolve_mt5_symbol(symbol) if symbol else None
|
||||
query = {"symbol": mt5_symbol} if mt5_symbol else {}
|
||||
r = self._request("GET", "/orders", params=query)
|
||||
@@ -749,34 +658,14 @@ class mt5bridge(ccxt.Exchange):
|
||||
return [self._parse_order(o, status="open") for o in orders]
|
||||
|
||||
def fetch_closed_orders(self, symbol=None, since=None, limit=None, params={}):
|
||||
"""Fetch closed orders via history.
|
||||
|
||||
MT5Bridge exposes history as deals, not orders. We approximate
|
||||
closed orders by reading the deal history and pairing entries
|
||||
with exits.
|
||||
"""
|
||||
# For simplicity, return my_trades; for proper order tracking
|
||||
# one would need to join deals with the order they belong to.
|
||||
return self.fetch_my_trades(symbol, since, limit, params)
|
||||
|
||||
def fetch_my_trades(self, symbol=None, since=None, limit=None, params={}):
|
||||
"""Fetch historical trades.
|
||||
|
||||
Args:
|
||||
symbol: optional CCXT symbol
|
||||
since: start timestamp in ms
|
||||
limit: not used (returns all in range)
|
||||
params: can include "date_from", "date_to"
|
||||
|
||||
Returns:
|
||||
list of CCXT trade dicts
|
||||
"""
|
||||
if "date_from" in params and "date_to" in params:
|
||||
date_from = params["date_from"]
|
||||
date_to = params["date_to"]
|
||||
else:
|
||||
if since is None:
|
||||
# Default: last 7 days
|
||||
since_ts = int((time.time() - 7 * 86400) * 1000)
|
||||
else:
|
||||
since_ts = int(since)
|
||||
@@ -826,7 +715,6 @@ class mt5bridge(ccxt.Exchange):
|
||||
return trades
|
||||
|
||||
def _parse_order(self, o, status="open"):
|
||||
"""Convert MT5Bridge order dict to CCXT order dict."""
|
||||
ts = self._parse_datetime(o.get("time"))
|
||||
ccxt_sym = self._resolve_ccxt_symbol(o.get("symbol", ""))
|
||||
mt5_type = int(o.get("type", 0))
|
||||
@@ -868,30 +756,6 @@ class mt5bridge(ccxt.Exchange):
|
||||
# ──────── Public API: trading ────────
|
||||
|
||||
def create_order(self, symbol, type, side, amount, price=None, params={}):
|
||||
"""Create a new order.
|
||||
|
||||
Args:
|
||||
symbol: CCXT symbol (e.g. "XAU/USD")
|
||||
type: "market" | "limit" | "stop" | "stop_limit"
|
||||
side: "buy" | "sell"
|
||||
amount: volume in lots (e.g. 0.01)
|
||||
price: required for limit/stop orders
|
||||
params: extra params:
|
||||
- sl: stop loss price (default 0)
|
||||
- tp: take profit price (default 0)
|
||||
- magic: magic number (default 0)
|
||||
- comment: order comment, max 31 chars
|
||||
- deviation: slippage in points (default 10)
|
||||
- check_first: if True, validate via /order/check first
|
||||
|
||||
Returns:
|
||||
dict: CCXT order
|
||||
|
||||
Raises:
|
||||
InvalidOrder: order was rejected by broker
|
||||
InsufficientFunds: retcode indicates not enough margin
|
||||
ArgumentsRequired: missing required argument
|
||||
"""
|
||||
if type not in ("market", "limit", "stop", "stop_limit"):
|
||||
raise InvalidOrder(
|
||||
f"Unsupported order type '{type}'. "
|
||||
@@ -921,7 +785,6 @@ class mt5bridge(ccxt.Exchange):
|
||||
"deviation": int(params.get("deviation", 10)),
|
||||
}
|
||||
|
||||
# Optional pre-check
|
||||
if params.get("check_first", False):
|
||||
check = self._request("POST", "/order/check", json_body=body)
|
||||
check_result = check.get("data", {})
|
||||
@@ -936,7 +799,6 @@ class mt5bridge(ccxt.Exchange):
|
||||
f"comment={check_result.get('comment', '')}"
|
||||
)
|
||||
|
||||
# Send order
|
||||
r = self._request("POST", "/order/send", json_body={"request": body})
|
||||
result = r.get("data", {})
|
||||
retcode = int(result.get("retcode", 0))
|
||||
@@ -953,7 +815,6 @@ class mt5bridge(ccxt.Exchange):
|
||||
return self._make_order_from_result(result, symbol, type, side, body)
|
||||
|
||||
def _make_order_from_result(self, result, symbol, type, side, body):
|
||||
"""Build a CCXT order dict from a successful order result."""
|
||||
ts = self.milliseconds()
|
||||
order_id = str(result.get("order", ""))
|
||||
return {
|
||||
@@ -980,22 +841,6 @@ class mt5bridge(ccxt.Exchange):
|
||||
}
|
||||
|
||||
def cancel_order(self, id, symbol=None, params={}):
|
||||
"""Cancel a pending order (async via MQL5 helper EA).
|
||||
|
||||
Note: MT5Bridge doesn't expose a synchronous cancel endpoint. This
|
||||
method sends a GlobalVariable command that the MQL5 helper EA
|
||||
(Mt5BridgeHelper.mq5) reads and acts on by calling OrderDelete().
|
||||
|
||||
For reliable cancellation, deploy the helper EA on a separate chart.
|
||||
|
||||
Args:
|
||||
id: order ticket (string or int)
|
||||
symbol: optional symbol (unused)
|
||||
params: extra params
|
||||
|
||||
Returns:
|
||||
dict: {"info": ..., "id": str, "status": "canceling"}
|
||||
"""
|
||||
ticket = int(id)
|
||||
result = self.mql5.send_cancel_command(ticket)
|
||||
return {
|
||||
@@ -1003,3 +848,114 @@ class mt5bridge(ccxt.Exchange):
|
||||
"id": str(id),
|
||||
"status": "canceling",
|
||||
}
|
||||
|
||||
# ─────────────────────────────────────────────────────────────────
|
||||
# Watching (polling-based implementations of CCXT watch_* methods)
|
||||
# ─────────────────────────────────────────────────────────────────
|
||||
#
|
||||
# MT5Bridge doesn't expose WebSocket or SSE. The watch_* methods
|
||||
# below poll the synchronous fetch_* endpoints until a new value
|
||||
# is detected, then return it. This is functionally equivalent to
|
||||
# a real watch for low-frequency use cases.
|
||||
#
|
||||
# For high-frequency real-time streaming, upgrade MT5Bridge to
|
||||
# support WebSocket (see README of this package).
|
||||
|
||||
async def watch_ticker(self, symbol, params={}):
|
||||
"""Watch the ticker for a symbol by polling.
|
||||
|
||||
Polls the /symbols/{symbol}/tick endpoint until a new tick is
|
||||
detected (timestamp changes, or bid/ask changes), then returns it.
|
||||
|
||||
Args:
|
||||
symbol: CCXT symbol
|
||||
params: extra params:
|
||||
- poll_interval_ms: polling interval in ms (default 500)
|
||||
- max_wait_ms: max time to wait before returning current
|
||||
value even if no change (default 30000)
|
||||
|
||||
Returns:
|
||||
dict: CCXT ticker
|
||||
|
||||
Example:
|
||||
>>> while True:
|
||||
... ticker = await exchange.watch_ticker("XAU/USD")
|
||||
... print(ticker["bid"], ticker["ask"])
|
||||
"""
|
||||
poll_interval_ms = params.get("poll_interval_ms", 500)
|
||||
max_wait_ms = params.get("max_wait_ms", 30000)
|
||||
last_signature = None
|
||||
loop = asyncio.get_event_loop()
|
||||
deadline = loop.time() + (max_wait_ms / 1000.0)
|
||||
|
||||
while True:
|
||||
ticker = await loop.run_in_executor(None, self.fetch_ticker, symbol)
|
||||
# Use (timestamp, bid, ask) as the change signature
|
||||
signature = (ticker.get("timestamp"), ticker.get("bid"), ticker.get("ask"))
|
||||
|
||||
if last_signature is None or signature != last_signature:
|
||||
last_signature = signature
|
||||
return ticker
|
||||
|
||||
# No change yet
|
||||
now = loop.time()
|
||||
if now >= deadline:
|
||||
return ticker # return current value, even if unchanged
|
||||
|
||||
await asyncio.sleep(poll_interval_ms / 1000.0)
|
||||
|
||||
async def watch_ohlcv(self, symbol, timeframe="1m", since=None, limit=None, params={}):
|
||||
"""Watch OHLCV for a symbol by polling.
|
||||
|
||||
Polls fetch_ohlcv until a new bar is detected (latest bar's
|
||||
timestamp changes), then returns the new bars.
|
||||
|
||||
Args:
|
||||
symbol: CCXT symbol
|
||||
timeframe: CCXT timeframe
|
||||
since: optional start timestamp in ms
|
||||
limit: number of bars to return (default 100)
|
||||
params: extra params:
|
||||
- poll_interval_ms: minimum polling interval in ms
|
||||
(default is auto-set per timeframe)
|
||||
- max_wait_ms: max time to wait before returning (default 600000)
|
||||
|
||||
Returns:
|
||||
list of [timestamp, open, high, low, close, volume]
|
||||
"""
|
||||
poll_interval_ms = params.get("poll_interval_ms",
|
||||
self.WATCH_OHLCV_POLL_MS.get(timeframe, 30000))
|
||||
max_wait_ms = params.get("max_wait_ms", 600000)
|
||||
last_bar_time = None
|
||||
loop = asyncio.get_event_loop()
|
||||
deadline = loop.time() + (max_wait_ms / 1000.0)
|
||||
fetch_limit = min(limit or 2, 5) # only need last few bars to detect new one
|
||||
|
||||
def _fetch():
|
||||
return self.fetch_ohlcv(symbol, timeframe, since=since, limit=fetch_limit, params=params)
|
||||
|
||||
while True:
|
||||
ohlcv = await loop.run_in_executor(None, _fetch)
|
||||
if ohlcv:
|
||||
latest_time = ohlcv[-1][0]
|
||||
if last_bar_time is None or latest_time != last_bar_time:
|
||||
last_bar_time = latest_time
|
||||
# Return full requested limit, not just the last few
|
||||
if limit and limit > fetch_limit:
|
||||
return await loop.run_in_executor(
|
||||
None,
|
||||
lambda: self.fetch_ohlcv(
|
||||
symbol, timeframe, since=since, limit=limit, params=params
|
||||
),
|
||||
)
|
||||
return ohlcv
|
||||
|
||||
now = loop.time()
|
||||
if now >= deadline:
|
||||
# Return what we have, even if no new bar
|
||||
if ohlcv:
|
||||
return ohlcv
|
||||
# No data at all; let the user see the empty result
|
||||
return ohlcv
|
||||
|
||||
await asyncio.sleep(poll_interval_ms / 1000.0)
|
||||
|
||||
Reference in New Issue
Block a user