195 lines
6.9 KiB
Python
195 lines
6.9 KiB
Python
"""
|
|
SymbolMetadataClient - Query symbol metadata from Iceberg.
|
|
|
|
Tickers use Nautilus format: "BTC/USDT.BINANCE" (market_id.exchange_id).
|
|
"""
|
|
|
|
import logging
|
|
from typing import Optional, Dict, NamedTuple
|
|
from pyiceberg.catalog import load_catalog
|
|
from pyiceberg.expressions import EqualTo, And
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
|
|
def format_ticker(exchange_id: str, market_id: str) -> str:
|
|
"""Format a ticker in Nautilus convention: 'BTC/USDT.BINANCE'."""
|
|
return f"{market_id}.{exchange_id}"
|
|
|
|
|
|
def parse_ticker(ticker: str) -> tuple[str, str]:
|
|
"""
|
|
Parse a Nautilus-format ticker into (exchange_id, market_id).
|
|
|
|
Args:
|
|
ticker: e.g. "BTC/USDT.BINANCE"
|
|
|
|
Returns:
|
|
(exchange_id, market_id) e.g. ("BINANCE", "BTC/USDT")
|
|
|
|
Raises:
|
|
ValueError: if the ticker does not contain a dot separator
|
|
"""
|
|
if "." not in ticker:
|
|
raise ValueError(
|
|
f"Invalid ticker format '{ticker}'. Expected Nautilus format: 'MARKET.EXCHANGE' "
|
|
f"(e.g., 'BTC/USDT.BINANCE')"
|
|
)
|
|
# Split on the LAST dot to handle market IDs that could theoretically contain dots
|
|
dot_pos = ticker.rfind(".")
|
|
market_id = ticker[:dot_pos]
|
|
exchange_id = ticker[dot_pos + 1:]
|
|
return exchange_id, market_id
|
|
|
|
|
|
class SymbolMetadata(NamedTuple):
|
|
"""Symbol metadata for Nautilus Instrument construction and order validation."""
|
|
exchange_id: str
|
|
market_id: str
|
|
market_type: Optional[str] = None
|
|
description: Optional[str] = None
|
|
base_asset: Optional[str] = None
|
|
quote_asset: Optional[str] = None
|
|
# Nautilus Instrument fields
|
|
price_precision: Optional[int] = None # decimal places for prices
|
|
size_precision: Optional[int] = None # decimal places for quantities
|
|
tick_size: Optional[float] = None # minimum price increment
|
|
lot_size: Optional[float] = None # minimum order size
|
|
min_notional: Optional[float] = None # minimum order value in quote currency
|
|
margin_init: Optional[float] = None # initial margin (futures/perps only)
|
|
margin_maint: Optional[float] = None # maintenance margin (futures/perps only)
|
|
maker_fee: Optional[float] = None # maker fee rate (e.g., 0.001 = 0.1%)
|
|
taker_fee: Optional[float] = None # taker fee rate
|
|
contract_multiplier: Optional[float] = None # for derivatives (default 1.0)
|
|
|
|
@property
|
|
def ticker(self) -> str:
|
|
"""Nautilus-format ticker: 'BTC/USDT.BINANCE'."""
|
|
return format_ticker(self.exchange_id, self.market_id)
|
|
|
|
|
|
class SymbolMetadataClient:
|
|
"""
|
|
Client for querying symbol metadata from Iceberg.
|
|
|
|
Tickers use Nautilus format: "BTC/USDT.BINANCE"
|
|
"""
|
|
|
|
def __init__(
|
|
self,
|
|
catalog_uri: str,
|
|
namespace: str = "trading",
|
|
s3_endpoint: Optional[str] = None,
|
|
s3_access_key: Optional[str] = None,
|
|
s3_secret_key: Optional[str] = None,
|
|
):
|
|
self.catalog_uri = catalog_uri
|
|
self.namespace = namespace
|
|
|
|
catalog_props = {"uri": catalog_uri}
|
|
if s3_endpoint:
|
|
catalog_props["s3.endpoint"] = s3_endpoint
|
|
catalog_props["s3.path-style-access"] = "true"
|
|
if s3_access_key:
|
|
catalog_props["s3.access-key-id"] = s3_access_key
|
|
if s3_secret_key:
|
|
catalog_props["s3.secret-access-key"] = s3_secret_key
|
|
|
|
self.catalog = load_catalog("trading", **catalog_props)
|
|
self._table = None
|
|
self._cache: Dict[str, SymbolMetadata] = {}
|
|
|
|
@property
|
|
def table(self):
|
|
if self._table is None:
|
|
try:
|
|
self._table = self.catalog.load_table(f"{self.namespace}.symbol_metadata")
|
|
log.info(f"Loaded symbol_metadata table from {self.namespace}")
|
|
except Exception as e:
|
|
raise RuntimeError(
|
|
f"Failed to load symbol_metadata table from {self.namespace}.symbol_metadata: {e}"
|
|
) from e
|
|
return self._table
|
|
|
|
def get_metadata(self, ticker: str) -> SymbolMetadata:
|
|
"""
|
|
Get metadata for a ticker (e.g., "BTC/USDT.BINANCE").
|
|
|
|
Args:
|
|
ticker: Market identifier in Nautilus format "MARKET.EXCHANGE"
|
|
|
|
Returns:
|
|
SymbolMetadata with Nautilus instrument fields
|
|
|
|
Raises:
|
|
ValueError: If ticker format is invalid or metadata not found
|
|
"""
|
|
if ticker in self._cache:
|
|
return self._cache[ticker]
|
|
|
|
exchange_id, market_id = parse_ticker(ticker)
|
|
|
|
try:
|
|
df = self.table.scan(
|
|
row_filter=And(
|
|
EqualTo("exchange_id", exchange_id),
|
|
EqualTo("market_id", market_id)
|
|
)
|
|
).to_pandas()
|
|
|
|
if df.empty:
|
|
raise ValueError(
|
|
f"No metadata found for ticker '{ticker}' "
|
|
f"(exchange_id='{exchange_id}', market_id='{market_id}'). "
|
|
f"The symbol may not be configured in the system."
|
|
)
|
|
|
|
if len(df) > 1:
|
|
log.warning(f"Multiple metadata entries found for {ticker}, using first entry")
|
|
|
|
row = df.iloc[0]
|
|
|
|
def _opt_int(col):
|
|
v = row.get(col)
|
|
return int(v) if v is not None and not (isinstance(v, float) and v != v) else None
|
|
|
|
def _opt_float(col):
|
|
v = row.get(col)
|
|
return float(v) if v is not None and not (isinstance(v, float) and v != v) else None
|
|
|
|
metadata = SymbolMetadata(
|
|
exchange_id=exchange_id,
|
|
market_id=market_id,
|
|
market_type=row.get("market_type"),
|
|
description=row.get("description"),
|
|
base_asset=row.get("base_asset"),
|
|
quote_asset=row.get("quote_asset"),
|
|
price_precision=_opt_int("price_precision"),
|
|
size_precision=_opt_int("size_precision"),
|
|
tick_size=_opt_float("tick_size"),
|
|
lot_size=_opt_float("lot_size"),
|
|
min_notional=_opt_float("min_notional"),
|
|
margin_init=_opt_float("margin_init"),
|
|
margin_maint=_opt_float("margin_maint"),
|
|
maker_fee=_opt_float("maker_fee"),
|
|
taker_fee=_opt_float("taker_fee"),
|
|
contract_multiplier=_opt_float("contract_multiplier"),
|
|
)
|
|
|
|
self._cache[ticker] = metadata
|
|
log.debug(
|
|
f"Loaded metadata for {ticker}: price_precision={metadata.price_precision}, "
|
|
f"tick_size={metadata.tick_size}, maker_fee={metadata.maker_fee}"
|
|
)
|
|
return metadata
|
|
|
|
except ValueError:
|
|
raise
|
|
except Exception as e:
|
|
raise RuntimeError(f"Failed to query metadata for ticker '{ticker}': {e}") from e
|
|
|
|
def clear_cache(self):
|
|
"""Clear the metadata cache (useful for testing or forcing reloads)."""
|
|
self._cache.clear()
|
|
log.info("Symbol metadata cache cleared")
|