пишем код ММ для ВТБ

Пишем безопасный исследовательский каркас: Python‑подобный код для симуляции/бэктеста логики ММ по ВТБ, с:

1. заполненными helper‑методами упрощённой математикой,
2. dataclass‑структурами для параметров, лимитов, фич, стакана, сделок и т.д.

Важно: это не боевой HFT/исполнительный код и не инструкция по реальной торговле. Здесь нет подключения к бирже, нет низколатентной логики, нет реального order manager. Это каркас для исследования, симуляции и обучения модели.

Ибо торговля нынче правратилась в обучение модели… Ну да то другое… Итак, поехали

Вариант 1. Python‑style каркас с заполненными helper‑методами  

from dataclasses import dataclass, field
from typing import List, Dict, Optional, Tuple
import math
import statistics

# =========================================================
# DATA STRUCTURES
# =========================================================

@dataclass
class BookLevel:
price: float
size: float

@dataclass
class OrderBook:
bids: List[BookLevel] # sorted desc by price
asks: List[BookLevel] # sorted asc by price
ts: float = 0.0

@dataclass
class Trade:
price: float
size: float
side: str # «BUY» or «SELL» from aggressor perspective
ts: float

@dataclass
class Fill:
symbol: str
side: str # our side: «BUY» or «SELL»
price: float
size: float
ts: float

@dataclass
class Order:
order_id: str
symbol: str
side: str
price: float
size: float
ts: float

@dataclass
class SessionInfo:
is_open_auction: bool = False
is_news_window: bool = False
is_closing_auction: bool = False

@dataclass
class LatencyStats:
feed_delay: float = 0.0
order_ack_delay: float = 0.0

@dataclass
class OptionSurface:
atm_iv: float = 0.30
put_call_skew: float = 0.0
gamma_exposure_proxy: float = 0.0
jump_proxy: float = 0.0

@dataclass
class Params:
# Fair value
k_stress_beta: float = 0.5
b_imoex: float = 0.8
b_banks: float = 0.7
b_fx: float = -0.2
b_oil: float = 0.05
b_momentum: float = 0.15
b_fut_mom: float = 0.5
b_basis: float = 0.2
b_fut_flow: float = 0.25

# Toxicity
t_qi: float = 0.4
t_mli: float = 0.3
t_cancel: float = 0.2
t_tradeflow: float = 0.35
t_thinning: float = 0.25
t_bidbreak: float = 0.5
t_forced: float = 0.6
t_futflow: float = 0.25

# Jump risk
j_gap: float = 0.4
j_event: float = 0.7
j_iv: float = 0.3
j_opt: float = 0.4
j_stress: float = 0.35

# Inventory / skew
inv_spot_penalty: float = 0.00002
inv_fut_penalty: float = 0.00001
s_ofi: float = 0.15
s_qi: float = 0.10
s_bidbreak: float = 0.20
s_forced: float = 0.25
s_inventory: float = 0.00003
s_stress: float = 0.10
s_futsignal: float = 0.12
s_toxicity: float = 0.08

# Spread
min_spread: float = 0.0002
max_spread: float = 0.01
sp_rv_short: float = 0.8
sp_rv_medium: float = 0.4
sp_toxicity: float = 0.6
sp_jump: float = 0.7
sp_event: float = 0.5
sp_basis: float = 0.2
sp_hedge: float = 0.2
sp_iv: float = 0.15
sp_latency: float = 0.5

# Size
base_size: float = 100000.0
min_size: float = 1000.0
max_size: float = 300000.0
sz_rv: float = 2.0
sz_toxicity: float = 1.5
sz_jump: float = 1.2
sz_forced: float = 1.8
sz_inventory: float = 0.00001

# Hedge / basis
hedge_cost_liq: float = 0.001
hedge_cost_basis: float = 0.0005
basis_risk_mult: float = 0.2
spot_to_fut_beta: float = 1.0
min_hedge_threshold: float = 5000.0
min_fut_liquidity: float = 0.2
forced_flow_urgency_threshold: float = 0.7

# Risk filters
event_mode_size_cap: float = 10000.0
extra_bid_step_on_break_risk: int = 2
break_risk_size_multiplier: float = 0.5
extra_bid_step_forced_flow: int = 3
extra_ask_step_forced_flow: int = 1
forced_flow_size_multiplier: float = 0.4
inventory_unwind_step: int = 2

# Markout
bad_markout_threshold: float = -0.0005
good_markout_threshold: float = 0.0005
markout_penalty_step: float = 0.05
markout_reward_step: float = 0.02
max_local_toxicity_penalty: float = 1.0

# Queue model
max_queue_ahead_bid: float = 500000.0
max_queue_ahead_ask: float = 500000.0
high_fill_prob: float = 0.7
high_bid_break_prob: float = 0.7
queue_toxicity_size_cut: float = 0.6
good_queue_fill_prob: float = 0.6

# Regimes
stress_threshold: float = 0.7
high_vol_threshold: float = 0.02
volatile_threshold: float = 0.01

regime_fv_volatile: float = 0.0
regime_fv_stressed: float = -0.0002
regime_fv_liq: float = -0.0005
regime_fv_event: float = -0.0003

inv_regime_normal: float = 0.00001
inv_regime_volatile: float = 0.00002
inv_regime_stressed: float = 0.00003
inv_regime_liq: float = 0.00005
inv_regime_event: float = 0.00004

skew_regime_volatile: float = -0.00005
skew_regime_stressed: float = -0.00010
skew_regime_liq: float = -0.00020
skew_regime_event: float = -0.00015

# Cross-impact
x_sber: float = 0.25
x_imoex_fut: float = 0.20
x_bank_index: float = 0.30
x_usdrub_cross: float = -0.10
x_skew_mult: float = 0.3

# Defensive mode
runaway_forced_flow_threshold: float = 0.9

@dataclass
class RiskLimits:
event_hard_limit: float = 0.85
forced_flow_limit: float = 0.75
bid_break_limit: float = 0.75
inventory_soft_limit: float = 300000.0
inventory_hard_limit: float = 700000.0
max_intraday_loss: float = -5_000_000.0
max_feed_delay: float = 0.5

@dataclass
class MarketFeatures:
mid_spot: float = 0.0
spread_spot: float = 0.0
qi: float = 0.0
mli: float = 0.0
cancel_pressure: float = 0.0
trade_sign_int: float = 0.0
book_thinning: float = 0.0
bid_break_prob: float = 0.0

rv_short: float = 0.0
rv_medium: float = 0.0
momentum_short: float = 0.0
gap_risk: float = 0.0

mid_fut: float = 0.0
fut_momentum: float = 0.0
fut_trade_sign: float = 0.0
basis: float = 0.0
basis_zscore: float = 0.0
fut_liquidity: float = 0.0

iv_atm: float = 0.0
iv_skew: float = 0.0
gamma_zone: float = 0.0
option_jump_proxy: float = 0.0

imoex_return: float = 0.0
bank_sector_ret: float = 0.0
usdrub_move: float = 0.0
oil_move: float = 0.0
market_stress: float = 0.0

event_risk: float = 0.0
forced_flow: float = 0.0

sber_return: float = 0.0
imoex_fut_return: float = 0.0
bank_index_return: float = 0.0
usdrub_cross_return: float = 0.0

queue_bid_ahead: float = 0.0
queue_ask_ahead: float = 0.0
queue_fill_prob_bid: float = 0.0
queue_fill_prob_ask: float = 0.0

# =========================================================
# SIMPLE MOCK APIS FOR RESEARCH / BACKTEST
# =========================================================

class MockLogger:
def info(self, msg): print("[INFO]", msg)
def warning(self, msg): print("[WARN]", msg)

class MockDataAPI:
def __init__(self):
self.reference = {«VTBR»: 0.025, «VTBR_FUT»: 0.0252}
self.future_mid_cache = {«VTBR»: {«short»: 0.025, «medium»: 0.025}}

def get_spot_order_book(self, symbol) -> OrderBook:
return OrderBook(
bids=[BookLevel(0.02498, 200000), BookLevel(0.02497, 300000), BookLevel(0.02496, 500000)],
asks=[BookLevel(0.02500, 180000), BookLevel(0.02501, 250000), BookLevel(0.02502, 400000)],
ts=0.0
)

def get_recent_spot_trades(self, symbol) -> List[Trade]:
return [
Trade(0.02499, 50000, «SELL», 1.0),
Trade(0.02498, 70000, «SELL», 2.0),
Trade(0.02500, 30000, «BUY», 3.0),
]

def get_futures_order_book(self, symbol) -> OrderBook:
return OrderBook(
bids=[BookLevel(0.02515, 100000), BookLevel(0.02514, 150000)],
asks=[BookLevel(0.02518, 90000), BookLevel(0.02519, 120000)],
ts=0.0
)

def get_recent_futures_trades(self, symbol) -> List[Trade]:
return [
Trade(0.02516, 40000, «SELL», 1.0),
Trade(0.02515, 60000, «SELL», 2.0),
Trade(0.02518, 20000, «BUY», 3.0),
]

def get_options_surface(self, symbol) -> OptionSurface:
return OptionSurface(atm_iv=0.42, put_call_skew=0.08, gamma_exposure_proxy=0.3, jump_proxy=0.35)

def get_market_factors(self) -> Dict:
return {
«imoex_return_short»: -0.008,
«bank_sector_return_short»: -0.012,
«usdrub_return_short»: 0.004,
«oil_return_short»: -0.003,
«cross_asset_vol»: 0.02,
}

def get_news_risk_flags(self, symbol) -> Dict:
return {«headline_risk»: 0.2, «issuer_risk»: 0.3, «macro_risk»: 0.4}

def get_latency_metrics(self) -> Dict:
return {«feed_delay»: 0.02, «order_ack_delay»: 0.01}

def get_session_state(self) -> SessionInfo:
return SessionInfo(is_open_auction=False, is_news_window=False, is_closing_auction=False)

def get_cross_asset_data(self, symbols: List[str]) -> Dict:
return {
«SBER»: {«return_short»: -0.010},
«IMOEX_FUT»: {«return_short»: -0.009},
«MOEXBANK»: {«return_short»: -0.011},
«USDRUB»: {«return_short»: 0.004},
}

def estimate_queue_ahead(self, active_order: Optional[Order], spot_book: OrderBook) -> float:
if active_order is None:
return 0.0
levels = spot_book.bids if active_order.side == «BUY» else spot_book.asks
for lvl in levels:
if abs(lvl.price — active_order.price) < 1e-12:
return lvl.size
return 0.0

def estimate_fill_probability(self, side: str, spot_book: OrderBook, spot_trades: List[Trade]) -> float:
top_size = spot_book.bids[0].size if side == «BUY» else spot_book.asks[0].size
traded = sum(t.size for t in spot_trades if ((side == «BUY» and t.side == «SELL») or (side == «SELL» and t.side == «BUY»)))
x = traded / max(top_size, 1.0)
return max(0.0, min(1.0, x))

def get_future_mid(self, symbol: str, horizon: str) -> float:
return self.future_mid_cache.get(symbol, {}).get(horizon, self.reference.get(symbol, 0.0))

def reference_price(self, symbol: str) -> float:
return self.reference.get(symbol, 0.0)

class MockExecutionAPI:
def __init__(self):
self.orders = {}
self.next_id = 1
self.pending_fills: List[Fill] = []

def place_or_replace_limit_order(self, side: str, symbol: str, price: float, size: float) -> Order:
oid = f«O{self.next_id}»
self.next_id += 1
order = Order(order_id=oid, symbol=symbol, side=side, price=price, size=size, ts=0.0)
self.orders[oid] = order
return order

def cancel_order(self, order: Optional[Order]):
if order and order.order_id in self.orders:
del self.orders[order.order_id]

def cancel_all_orders(self):
self.orders = {}

def send_or_adjust_futures_hedge_order(self, symbol: str, hedge_delta: float, urgency: str):
# Research stub: no real execution
pass

def get_new_fills(self, symbols: List[str]) -> List[Fill]:
fills = self.pending_fills[:]
self.pending_fills = []
return fills

# =========================================================
# MARKET MAKER CLASS
# =========================================================

class MarketMakerVTB:
def __init__(self, params: Params, risk_limits: RiskLimits, data_api, execution_api, logger):
self.params = params
self.risk_limits = risk_limits
self.data_api = data_api
self.execution_api = execution_api
self.logger = logger

self.inventory_spot = 0.0
self.inventory_fut = 0.0
self.pnl = 0.0

self.active_bid_order: Optional[Order] = None
self.active_ask_order: Optional[Order] = None

self.local_toxicity_penalty = 0.0
self.markout_history = []
self.fill_history = []
self.regime_state = «normal»

self.rolling_stats = self._init_rolling_stats()
self.enabled = True

# — MAIN LOOP ----------------

def on_market_update(self):
if not self.enabled:
return

market = self._ingest_data()
features = self._build_features(market)

regime = self.classify_regime(features)
self.regime_state = regime

cross_impact = self.compute_cross_impact(features)
fair_value = self.compute_fair_value(features, regime, cross_impact)
toxicity = self.compute_toxicity(features, regime)
jump_risk = self.compute_jump_risk(features, regime)

hedge_cost = self.estimate_hedge_cost(features)
basis_risk = self.estimate_basis_risk(features)

reservation_price = self.compute_reservation_price(fair_value, regime)
skew = self.compute_skew(features, regime, toxicity, cross_impact)
spread = self.compute_spread(features, regime, toxicity, jump_risk, hedge_cost, basis_risk)
quote_size = self.compute_quote_size(features, regime, toxicity, jump_risk)

bid_px, ask_px = self.compute_preliminary_quotes(reservation_price, skew, spread)

bid_px, ask_px, quote_size = self.apply_queue_position_adjustments(
bid_px, ask_px, quote_size, features, regime
)

bid_px, ask_px, quote_size = self.apply_risk_filters(
bid_px, ask_px, quote_size, features, regime
)

self.hedge_with_futures_if_needed(features, regime)
self.manage_orders(bid_px, ask_px, quote_size, features, regime)
self.process_fills_and_feedback(features, regime)
self.run_kill_switch_checks(features, regime)

self.logger.info(
f«regime={regime} fv={fair_value:.6f} bid={bid_px:.6f} ask={ask_px:.6f} „
f“size={quote_size:.0f} inv={self.inventory_spot:.0f} tox={toxicity:.3f}»
)

# — DATA ----------------

def _ingest_data(self):
return {
«spot_book»: self.data_api.get_spot_order_book(«VTBR»),
«spot_trades»: self.data_api.get_recent_spot_trades(«VTBR»),
«fut_book»: self.data_api.get_futures_order_book(«VTBR_FUT»),
«fut_trades»: self.data_api.get_recent_futures_trades(«VTBR_FUT»),
«options_data»: self.data_api.get_options_surface(«VTBR»),
«market_factors»: self.data_api.get_market_factors(),
«news_flags»: self.data_api.get_news_risk_flags(«VTBR»),
«latency_stats»: self.data_api.get_latency_metrics(),
«session_info»: self.data_api.get_session_state(),
«cross_assets»: self.data_api.get_cross_asset_data(
symbols=[«SBER», «IMOEX_FUT», «MOEXBANK», «USDRUB»]
),
}

def _build_features(self, market) -> MarketFeatures:
f = MarketFeatures()

spot_book = market[«spot_book»]
spot_trades = market[«spot_trades»]
fut_book = market[«fut_book»]
fut_trades = market[«fut_trades»]
options_data = market[«options_data»]
mf = market[«market_factors»]
news = market[«news_flags»]
cross_assets = market[«cross_assets»]

f.mid_spot = self.compute_mid(spot_book)
f.spread_spot = self.compute_spread_from_book(spot_book)
f.qi = self.compute_queue_imbalance(spot_book, level=1)
f.mli = self.compute_multilevel_imbalance(spot_book, levels=5)
f.cancel_pressure = self.compute_cancel_pressure(spot_book)
f.trade_sign_int = self.compute_trade_sign_intensity(spot_trades)
f.book_thinning = self.compute_book_thinning(spot_book)
f.bid_break_prob = self.estimate_bid_break_probability(spot_book, spot_trades)

f.rv_short = self.realized_volatility(«VTBR», window=«short»)
f.rv_medium = self.realized_volatility(«VTBR», window=«medium»)
f.momentum_short = self.intraday_momentum(«VTBR», window=«short»)
f.gap_risk = self.estimate_gap_risk(market[«session_info»], news, mf)

f.mid_fut = self.compute_mid(fut_book)
f.fut_momentum = self.intraday_momentum(«VTBR_FUT», window=«short»)
f.fut_trade_sign = self.compute_trade_sign_intensity(fut_trades)
f.basis = self.compute_basis(f.mid_spot, f.mid_fut)
f.basis_zscore = self.zscore(«basis», f.basis)
f.fut_liquidity = self.estimate_liquidity(fut_book, fut_trades)

f.iv_atm = self.get_atm_implied_vol(options_data)
f.iv_skew = self.get_put_call_skew(options_data)
f.gamma_zone = self.estimate_gamma_zone(options_data, f.mid_spot)
f.option_jump_proxy = self.estimate_option_jump_risk(options_data)

f.imoex_return = mf[«imoex_return_short»]
f.bank_sector_ret = mf[«bank_sector_return_short»]
f.usdrub_move = mf[«usdrub_return_short»]
f.oil_move = mf[«oil_return_short»]
f.market_stress = self.estimate_market_stress(mf)

f.event_risk = self.estimate_event_risk(news)
f.forced_flow = self.estimate_forced_flow(spot_trades, fut_trades, spot_book, fut_book)

f.sber_return = cross_assets[«SBER»][«return_short»]
f.imoex_fut_return = cross_assets[«IMOEX_FUT»][«return_short»]
f.bank_index_return = cross_assets[«MOEXBANK»][«return_short»]
f.usdrub_cross_return = cross_assets[«USDRUB»][«return_short»]

f.queue_bid_ahead = self.estimate_queue_ahead(«BUY», self.active_bid_order, spot_book)
f.queue_ask_ahead = self.estimate_queue_ahead(«SELL», self.active_ask_order, spot_book)
f.queue_fill_prob_bid = self.estimate_fill_probability(«BUY», spot_book, spot_trades)
f.queue_fill_prob_ask = self.estimate_fill_probability(«SELL», spot_book, spot_trades)

return f

# — CORE MODELS ----------------

def compute_fair_value(self, f: MarketFeatures, regime: str, cross_impact: float) -> float:
stress_mult = 1.0 + self.params.k_stress_beta * f.market_stress

fv_slow = (
self.params.b_imoex * f.imoex_return
+ self.params.b_banks * f.bank_sector_ret
+ self.params.b_fx * f.usdrub_move
+ self.params.b_oil * f.oil_move
+ self.params.b_momentum * f.momentum_short
)

fv_fast = (
self.params.b_fut_mom * f.fut_momentum
+ self.params.b_basis * f.basis_zscore
+ self.params.b_fut_flow * f.fut_trade_sign
)

return (
self.reference_price(«VTBR»)
+ stress_mult * (fv_slow + fv_fast)
+ cross_impact
+ self.regime_fair_value_adjustment(regime, f)
)

def compute_toxicity(self, f: MarketFeatures, regime: str) -> float:
tox = (
self.params.t_qi * self.negative_part(f.qi)
+ self.params.t_mli * self.negative_part(f.mli)
+ self.params.t_cancel * self.positive_part(f.cancel_pressure)
+ self.params.t_tradeflow * self.negative_part(f.trade_sign_int)
+ self.params.t_thinning * f.book_thinning
+ self.params.t_bidbreak * f.bid_break_prob
+ self.params.t_forced * f.forced_flow
+ self.params.t_futflow * self.negative_part(f.fut_trade_sign)
+ self.local_toxicity_penalty
)
return tox * self.regime_toxicity_multiplier(regime)

def compute_jump_risk(self, f: MarketFeatures, regime: str) -> float:
jr = (
self.params.j_gap * f.gap_risk
+ self.params.j_event * f.event_risk
+ self.params.j_iv * f.iv_atm
+ self.params.j_opt * f.option_jump_proxy
+ self.params.j_stress * f.market_stress
)
return jr * self.regime_jump_multiplier(regime)

def compute_reservation_price(self, fair_value: float, regime: str) -> float:
inv_penalty = (
self.params.inv_spot_penalty * self.inventory_spot
+ self.params.inv_fut_penalty * self.inventory_fut
)
regime_penalty = self.regime_inventory_penalty(regime) * self.inventory_spot
return fair_value — inv_penalty — regime_penalty

def compute_skew(self, f: MarketFeatures, regime: str, toxicity: float, cross_impact: float) -> float:
skew = (
— self.params.s_ofi * self.negative_part(f.trade_sign_int)
— self.params.s_qi * self.negative_part(f.qi)
— self.params.s_bidbreak * f.bid_break_prob
— self.params.s_forced * f.forced_flow
— self.params.s_inventory * self.inventory_spot
— self.params.s_stress * f.market_stress
— self.params.s_futsignal * self.negative_part(f.fut_trade_sign)
)
skew += self.regime_skew_adjustment(regime, f)
skew += self.cross_impact_skew_adjustment(cross_impact)
skew -= self.params.s_toxicity * toxicity
return skew

def compute_spread(self, f: MarketFeatures, regime: str, toxicity: float, jump_risk: float,
hedge_cost: float, basis_risk: float) -> float:
spread = (
self.params.min_spread
+ self.params.sp_rv_short * f.rv_short
+ self.params.sp_rv_medium * f.rv_medium
+ self.params.sp_toxicity * toxicity
+ self.params.sp_jump * jump_risk
+ self.params.sp_event * f.event_risk
+ self.params.sp_basis * basis_risk
+ self.params.sp_hedge * hedge_cost
+ self.params.sp_iv * f.iv_atm
+ self.params.sp_latency * self.estimate_latency_risk(f)
)
spread *= self.regime_spread_multiplier(regime)
return self.clamp(spread, self.params.min_spread, self.params.max_spread)

def compute_quote_size(self, f: MarketFeatures, regime: str, toxicity: float, jump_risk: float) -> float:
denom = (
1.0
+ self.params.sz_rv * f.rv_short
+ self.params.sz_toxicity * toxicity
+ self.params.sz_jump * jump_risk
+ self.params.sz_forced * f.forced_flow
+ self.params.sz_inventory * abs(self.inventory_spot)
)
size = self.params.base_size / max(denom, 1e-9)
size *= self.regime_size_multiplier(regime)
return self.clamp(size, self.params.min_size, self.params.max_size)

def compute_preliminary_quotes(self, reservation_price: float, skew: float, spread: float) -> Tuple[float, float]:
bid_px = reservation_price + skew — spread / 2.0
ask_px = reservation_price + skew + spread / 2.0
bid_px = self.round_to_tick(bid_px, «VTBR»)
ask_px = self.round_to_tick(ask_px, «VTBR»)
if bid_px >= ask_px:
ask_px = bid_px + self.tick_size(«VTBR»)
return bid_px, ask_px

# — 6.1 MARKOUT MODEL ----------------

def process_fills_and_feedback(self, features: MarketFeatures, regime: str):
fills = self.execution_api.get_new_fills([«VTBR», «VTBR_FUT»])
for fill in fills:
self.update_inventory_and_pnl(fill)
self.fill_history.append(fill)

markout = self.compute_markout(fill, horizons=[«short», «medium»])
self.markout_history.append(markout)
self.update_markout_model(fill, markout, features, regime)

def compute_markout(self, fill: Fill, horizons: List[str]) -> Dict[str, float]:
result = {}
for h in horizons:
future_mid = self.data_api.get_future_mid(fill.symbol, h)
if fill.side == «BUY»:
result[h] = future_mid — fill.price
else:
result[h] = fill.price — future_mid
return result

def update_markout_model(self, fill: Fill, markout: Dict[str, float], features: MarketFeatures, regime: str):
short_markout = markout.get(«short», 0.0)

if short_markout < self.params.bad_markout_threshold:
self.local_toxicity_penalty += self.params.markout_penalty_step
elif short_markout > self.params.good_markout_threshold:
self.local_toxicity_penalty -= self.params.markout_reward_step

self.local_toxicity_penalty = self.clamp(
self.local_toxicity_penalty, 0.0, self.params.max_local_toxicity_penalty
)

# — 6.2 QUEUE POSITION MODEL ----------------

def apply_queue_position_adjustments(self, bid_px: float, ask_px: float, quote_size: float,
f: MarketFeatures, regime: str):
if f.queue_bid_ahead > self.params.max_queue_ahead_bid:
bid_px -= self.tick_size(«VTBR»)

if f.queue_ask_ahead > self.params.max_queue_ahead_ask:
ask_px += self.tick_size(«VTBR»)

if f.queue_fill_prob_bid > self.params.high_fill_prob and f.bid_break_prob > self.params.high_bid_break_prob:
quote_size *= self.params.queue_toxicity_size_cut

if regime == «normal» and f.queue_fill_prob_bid > self.params.good_queue_fill_prob:
pass

return bid_px, ask_px, self.clamp(quote_size, self.params.min_size, self.params.max_size)

def estimate_queue_ahead(self, side: str, active_order: Optional[Order], spot_book: OrderBook) -> float:
return self.data_api.estimate_queue_ahead(active_order, spot_book)

def estimate_fill_probability(self, side: str, spot_book: OrderBook, spot_trades: List[Trade]) -> float:
return self.data_api.estimate_fill_probability(side, spot_book, spot_trades)

# — 6.3 REGIME CLASSIFIER ----------------

def classify_regime(self, f: MarketFeatures) -> str:
if f.event_risk > self.risk_limits.event_hard_limit:
return «event»

if f.forced_flow > self.risk_limits.forced_flow_limit and f.bid_break_prob > self.risk_limits.bid_break_limit:
return «liquidation_cascade»

if f.market_stress > self.params.stress_threshold or f.rv_short > self.params.high_vol_threshold:
return «stressed»

if f.rv_short > self.params.volatile_threshold:
return «volatile»

return «normal»

def regime_fair_value_adjustment(self, regime: str, f: MarketFeatures) -> float:
return {
«normal»: 0.0,
«volatile»: self.params.regime_fv_volatile,
«stressed»: self.params.regime_fv_stressed,
«liquidation_cascade»: self.params.regime_fv_liq,
«event»: self.params.regime_fv_event,
}.get(regime, 0.0)

def regime_toxicity_multiplier(self, regime: str) -> float:
return {
«normal»: 1.0,
«volatile»: 1.2,
«stressed»: 1.5,
«liquidation_cascade»: 2.0,
«event»: 1.8,
}.get(regime, 1.0)

def regime_jump_multiplier(self, regime: str) -> float:
return {
«normal»: 1.0,
«volatile»: 1.1,
«stressed»: 1.4,
«liquidation_cascade»: 1.7,
«event»: 2.0,
}.get(regime, 1.0)

def regime_inventory_penalty(self, regime: str) -> float:
return {
«normal»: self.params.inv_regime_normal,
«volatile»: self.params.inv_regime_volatile,
«stressed»: self.params.inv_regime_stressed,
«liquidation_cascade»: self.params.inv_regime_liq,
«event»: self.params.inv_regime_event,
}.get(regime, self.params.inv_regime_normal)

def regime_skew_adjustment(self, regime: str, f: MarketFeatures) -> float:
return {
«normal»: 0.0,
«volatile»: self.params.skew_regime_volatile,
«stressed»: self.params.skew_regime_stressed,
«liquidation_cascade»: self.params.skew_regime_liq,
«event»: self.params.skew_regime_event,
}.get(regime, 0.0)

def regime_spread_multiplier(self, regime: str) -> float:
return {
«normal»: 1.0,
«volatile»: 1.2,
«stressed»: 1.5,
«liquidation_cascade»: 2.0,
«event»: 2.2,
}.get(regime, 1.0)

def regime_size_multiplier(self, regime: str) -> float:
return {
«normal»: 1.0,
«volatile»: 0.8,
«stressed»: 0.6,
«liquidation_cascade»: 0.35,
«event»: 0.3,
}.get(regime, 1.0)

# — 6.4 CROSS-IMPACT MODEL ----------------

def compute_cross_impact(self, f: MarketFeatures) -> float:
return (
self.params.x_sber * f.sber_return
+ self.params.x_imoex_fut * f.imoex_fut_return
+ self.params.x_bank_index * f.bank_index_return
+ self.params.x_usdrub_cross * f.usdrub_cross_return
)

def cross_impact_skew_adjustment(self, cross_impact: float) -> float:
return self.params.x_skew_mult * cross_impact

# — RISK / HEDGE / ORDERS ----------------

def apply_risk_filters(self, bid_px: float, ask_px: float, quote_size: float,
f: MarketFeatures, regime: str):
if f.event_risk > self.risk_limits.event_hard_limit:
quote_size = min(quote_size, self.params.event_mode_size_cap)

if f.bid_break_prob > self.risk_limits.bid_break_limit:
bid_px -= self.params.extra_bid_step_on_break_risk * self.tick_size(«VTBR»)
quote_size *= self.params.break_risk_size_multiplier

if f.forced_flow > self.risk_limits.forced_flow_limit:
bid_px -= self.params.extra_bid_step_forced_flow * self.tick_size(«VTBR»)
ask_px -= self.params.extra_ask_step_forced_flow * self.tick_size(«VTBR»)
quote_size *= self.params.forced_flow_size_multiplier

if abs(self.inventory_spot) > self.risk_limits.inventory_soft_limit:
step = self.params.inventory_unwind_step * self.tick_size(«VTBR»)
if self.inventory_spot > 0:
bid_px -= step
ask_px -= step
else:
bid_px += step
ask_px += step

quote_size = self.clamp(quote_size, self.params.min_size, self.params.max_size)
return bid_px, ask_px, quote_size

def hedge_with_futures_if_needed(self, f: MarketFeatures, regime: str):
desired_fut_hedge = self.compute_desired_futures_hedge(self.inventory_spot, f.basis)
hedge_delta = desired_fut_hedge — self.inventory_fut

if abs(hedge_delta) > self.params.min_hedge_threshold and f.fut_liquidity > self.params.min_fut_liquidity:
self.execution_api.send_or_adjust_futures_hedge_order(
symbol=«VTBR_FUT»,
hedge_delta=hedge_delta,
urgency=self.hedge_urgency(f, regime)
)

def manage_orders(self, bid_px: float, ask_px: float, quote_size: float, f: MarketFeatures, regime: str):
if self.should_cancel_bid(self.active_bid_order, bid_px, quote_size, f):
self.execution_api.cancel_order(self.active_bid_order)
self.active_bid_order = None

if self.should_cancel_ask(self.active_ask_order, ask_px, quote_size, f):
self.execution_api.cancel_order(self.active_ask_order)
self.active_ask_order = None

if self.can_quote_bid(f, regime):
self.active_bid_order = self.execution_api.place_or_replace_limit_order(
side=«BUY», symbol=«VTBR», price=bid_px, size=quote_size
)

if self.can_quote_ask(f, regime):
self.active_ask_order = self.execution_api.place_or_replace_limit_order(
side=«SELL», symbol=«VTBR», price=ask_px, size=quote_size
)

def run_kill_switch_checks(self, f: MarketFeatures, regime: str):
latency_stats = self.data_api.get_latency_metrics()

if self.pnl < self.risk_limits.max_intraday_loss:
self.disable_strategy(«loss_limit»)

if latency_stats[«feed_delay»] > self.risk_limits.max_feed_delay:
self.disable_strategy(«market_data_latency»)

if self.detect_data_inconsistency():
self.disable_strategy(«data_integrity»)

if self.detect_runaway_market(f, regime):
self.switch_to_extreme_defensive_mode()

# — SUPPORT ----------------

def update_inventory_and_pnl(self, fill: Fill):
if fill.symbol == «VTBR»:
if fill.side == «BUY»:
self.inventory_spot += fill.size
self.pnl -= fill.price * fill.size
else:
self.inventory_spot -= fill.size
self.pnl += fill.price * fill.size
elif fill.symbol == «VTBR_FUT»:
if fill.side == «BUY»:
self.inventory_fut += fill.size
else:
self.inventory_fut -= fill.size

def estimate_hedge_cost(self, f: MarketFeatures) -> float:
return (
self.params.hedge_cost_liq * (1.0 / max(f.fut_liquidity, 1e-9))
+ self.params.hedge_cost_basis * abs(f.basis_zscore)
)

def estimate_basis_risk(self, f: MarketFeatures) -> float:
return self.params.basis_risk_mult * abs(f.basis_zscore)

def estimate_latency_risk(self, f: MarketFeatures) -> float:
latency = self.data_api.get_latency_metrics()
return latency[«feed_delay»] * f.bid_break_prob

def compute_desired_futures_hedge(self, inventory_spot: float, basis: float) -> float:
return — self.params.spot_to_fut_beta * inventory_spot

def hedge_urgency(self, f: MarketFeatures, regime: str) -> str:
if regime in («liquidation_cascade», «event»):
return «high»
if f.forced_flow > self.params.forced_flow_urgency_threshold:
return «high»
return «normal»

def should_cancel_bid(self, active_order: Optional[Order], new_bid_px: float, new_size: float, f: MarketFeatures) -> bool:
if active_order is None:
return False
if abs(active_order.price — new_bid_px) > 1e-12:
return True
if f.bid_break_prob > self.risk_limits.bid_break_limit:
return True
return False

def should_cancel_ask(self, active_order: Optional[Order], new_ask_px: float, new_size: float, f: MarketFeatures) -> bool:
if active_order is None:
return False
if abs(active_order.price — new_ask_px) > 1e-12:
return True
return False

def can_quote_bid(self, f: MarketFeatures, regime: str) -> bool:
if abs(self.inventory_spot) > self.risk_limits.inventory_hard_limit and self.inventory_spot > 0:
return False
if regime == «event» and f.event_risk > self.risk_limits.event_hard_limit:
return False
return True

def can_quote_ask(self, f: MarketFeatures, regime: str) -> bool:
if abs(self.inventory_spot) > self.risk_limits.inventory_hard_limit and self.inventory_spot < 0:
return False
return True

def disable_strategy(self, reason: str):
self.execution_api.cancel_all_orders()
self.enabled = False
self.logger.warning(f«Strategy disabled: {reason}»)

def switch_to_extreme_defensive_mode(self):
self.params.base_size *= 0.25
self.params.min_spread *= 1.5

def detect_data_inconsistency(self) -> bool:
return False

def detect_runaway_market(self, f: MarketFeatures, regime: str) -> bool:
return regime == «liquidation_cascade» and f.forced_flow > self.params.runaway_forced_flow_threshold

# — NUMERICAL HELPERS ----------------

def compute_mid(self, book: OrderBook) -> float:
if not book.bids or not book.asks:
return 0.0
return 0.5 * (book.bids[0].price + book.asks[0].price)

def compute_spread_from_book(self, book: OrderBook) -> float:
if not book.bids or not book.asks:
return 0.0
return max(0.0, book.asks[0].price — book.bids[0].price)

def compute_queue_imbalance(self, book: OrderBook, level: int = 1) -> float:
n = min(level, len(book.bids), len(book.asks))
bid_sz = sum(x.size for x in book.bids[:n])
ask_sz = sum(x.size for x in book.asks[:n])
return (bid_sz — ask_sz) / max(bid_sz + ask_sz, 1e-9)

def compute_multilevel_imbalance(self, book: OrderBook, levels: int = 5) -> float:
n = min(levels, len(book.bids), len(book.asks))
bid_sz = sum(x.size for x in book.bids[:n])
ask_sz = sum(x.size for x in book.asks[:n])
return (bid_sz — ask_sz) / max(bid_sz + ask_sz, 1e-9)

def compute_cancel_pressure(self, book: OrderBook) -> float:
# Research proxy: thin bid relative to ask => positive cancel pressure
bid_sz = sum(x.size for x in book.bids[:3])
ask_sz = sum(x.size for x in book.asks[:3])
return (ask_sz — bid_sz) / max(ask_sz + bid_sz, 1e-9)

def compute_trade_sign_intensity(self, trades: List[Trade]) -> float:
signed = 0.0
total = 0.0
for t in trades:
sign = 1.0 if t.side == «BUY» else -1.0
signed += sign * t.size
total += t.size
return signed / max(total, 1e-9)

def compute_book_thinning(self, book: OrderBook) -> float:
top_bid = sum(x.size for x in book.bids[:1])
deep_bid = sum(x.size for x in book.bids[:3])
if deep_bid <= 0:
return 1.0
ratio = top_bid / deep_bid
return 1.0 — ratio

def estimate_bid_break_probability(self, book: OrderBook, trades: List[Trade]) -> float:
sell_flow = sum(t.size for t in trades if t.side == «SELL»)
top_bid = book.bids[0].size if book.bids else 1.0
x = sell_flow / max(top_bid, 1.0)
return self.sigmoid(2.0 * (x — 1.0))

def realized_volatility(self, symbol: str, window: str) -> float:
vals = self.rolling_stats.get(f«ret_{symbol}_{window}», [0.001, -0.0015, 0.0008, -0.0006])
if len(vals) < 2:
return 0.0
return statistics.pstdev(vals)

def intraday_momentum(self, symbol: str, window: str) -> float:
vals = self.rolling_stats.get(f«ret_{symbol}_{window}», [0.001, -0.0015, 0.0008, -0.0006])
return sum(vals)

def estimate_gap_risk(self, session_info: SessionInfo, news: Dict, market_factors: Dict) -> float:
base = 0.2 * news.get(«headline_risk», 0.0) + 0.2 * news.get(«macro_risk», 0.0)
if session_info.is_open_auction or session_info.is_closing_auction:
base += 0.2
return self.clamp(base, 0.0, 1.0)

def compute_basis(self, mid_spot: float, mid_fut: float) -> float:
if mid_spot <= 0:
return 0.0
return (mid_fut — mid_spot) / mid_spot

def zscore(self, key: str, value: float) -> float:
hist = self.rolling_stats.get(key, [0.0, 0.0005, -0.0003, 0.0002, -0.0001])
mu = sum(hist) / len(hist)
sd = statistics.pstdev(hist) if len(hist) > 1 else 1.0
sd = max(sd, 1e-9)
return (value — mu) / sd

def estimate_liquidity(self, book: OrderBook, trades: List[Trade]) -> float:
depth = sum(x.size for x in book.bids[:2]) + sum(x.size for x in book.asks[:2])
flow = sum(t.size for t in trades)
x = depth / max(flow, 1.0)
return self.clamp(x / 10.0, 0.0, 1.0)

def get_atm_implied_vol(self, options_data: OptionSurface) -> float:
return options_data.atm_iv

def get_put_call_skew(self, options_data: OptionSurface) -> float:
return options_data.put_call_skew

def estimate_gamma_zone(self, options_data: OptionSurface, spot: float) -> float:
return options_data.gamma_exposure_proxy

def estimate_option_jump_risk(self, options_data: OptionSurface) -> float:
return options_data.jump_proxy

def estimate_market_stress(self, market_factors: Dict) -> float:
x = (
abs(market_factors.get(«imoex_return_short», 0.0)) * 20
+ abs(market_factors.get(«bank_sector_return_short», 0.0)) * 20
+ abs(market_factors.get(«usdrub_return_short», 0.0)) * 10
)
return self.clamp(x, 0.0, 1.0)

def estimate_event_risk(self, news_flags: Dict) -> float:
x = (
0.4 * news_flags.get(«headline_risk», 0.0)
+ 0.3 * news_flags.get(«issuer_risk», 0.0)
+ 0.3 * news_flags.get(«macro_risk», 0.0)
)
return self.clamp(x, 0.0, 1.0)

def estimate_forced_flow(self, spot_trades: List[Trade], fut_trades: List[Trade],
spot_book: OrderBook, fut_book: OrderBook) -> float:
spot_sell = sum(t.size for t in spot_trades if t.side == «SELL»)
spot_buy = sum(t.size for t in spot_trades if t.side == «BUY»)
fut_sell = sum(t.size for t in fut_trades if t.side == «SELL»)
fut_buy = sum(t.size for t in fut_trades if t.side == «BUY»)

spot_imb = (spot_sell — spot_buy) / max(spot_sell + spot_buy, 1.0)
fut_imb = (fut_sell — fut_buy) / max(fut_sell + fut_buy, 1.0)

top_bid = spot_book.bids[0].size if spot_book.bids else 1.0
sell_pressure = spot_sell / max(top_bid, 1.0)

x = 0.4 * max(0.0, spot_imb) + 0.3 * max(0.0, fut_imb) + 0.3 * self.sigmoid(sell_pressure — 1.0)
return self.clamp(x, 0.0, 1.0)

def reference_price(self, symbol: str) -> float:
return self.data_api.reference_price(symbol)

def round_to_tick(self, px: float, symbol: str) -> float:
tick = self.tick_size(symbol)
return round(px / tick) * tick

def tick_size(self, symbol: str) -> float:
return 0.00001

def clamp(self, x: float, lo: float, hi: float) -> float:
return max(lo, min(hi, x))

def positive_part(self, x: float) -> float:
return max(0.0, x)

def negative_part(self, x: float) -> float:
return max(0.0, -x)

def sigmoid(self, x: float) -> float:
return 1.0 / (1.0 + math.exp(-x))

def _init_rolling_stats(self) -> Dict[str, List[float]]:
return {
«basis»: [0.0002, 0.0001, -0.0001, 0.0003, 0.0000],
«ret_VTBR_short»: [0.001, -0.0015, 0.0008, -0.0006],
«ret_VTBR_medium»: [0.002, -0.001, 0.0012, -0.0008, 0.0005],
«ret_VTBR_FUT_short»: [0.0012, -0.0017, 0.0009, -0.0007],
}

Вариант 2. Dataclass‑структуры как аккуратный каркас проекта

Ниже — более “проектный” вариант: как это можно разложить по сущностям, чтобы код был чище и удобнее для исследований.

1. Структуры конфигурации и состояния

<code class="language-python">from dataclasses import dataclass, field
from typing import List, Dict, Optional


@dataclass
class StrategyState:
    enabled: bool = True
    regime: str = "normal"

    inventory_spot: float = 0.0
    inventory_fut: float = 0.0
    pnl: float = 0.0

    local_toxicity_penalty: float = 0.0

    active_bid_order_id: Optional[str] = None
    active_ask_order_id: Optional[str] = None

    markout_history: List[Dict[str, float]] = field(default_factory=list)
    fill_history: List["Fill"] = field(default_factory=list)


@dataclass
class QuoteDecision:
    fair_value: float
    reservation_price: float
    skew: float
    spread: float
    bid_px: float
    ask_px: float
    quote_size: float
    toxicity: float
    jump_risk: float
    cross_impact: float
    regime: str


@dataclass
class MarketSnapshot:
    spot_book: "OrderBook"
    spot_trades: List["Trade"]
    fut_book: "OrderBook"
    fut_trades: List["Trade"]
    options_data: "OptionSurface"
    market_factors: Dict
    news_flags: Dict
    latency_stats: Dict
    session_info: "SessionInfo"
    cross_assets: Dict
</code>

Copy2. Структуры рыночных данных

<code class="language-python">@dataclass
class BookLevel:
    price: float
    size: float


@dataclass
class OrderBook:
    bids: List[BookLevel]
    asks: List[BookLevel]
    ts: float = 0.0


@dataclass
class Trade:
    price: float
    size: float
    side: str
    ts: float


@dataclass
class Fill:
    symbol: str
    side: str
    price: float
    size: float
    ts: float


@dataclass
class Order:
    order_id: str
    symbol: str
    side: str
    price: float
    size: float
    ts: float


@dataclass
class SessionInfo:
    is_open_auction: bool = False
    is_news_window: bool = False
    is_closing_auction: bool = False


@dataclass
class OptionSurface:
    atm_iv: float = 0.0
    put_call_skew: float = 0.0
    gamma_exposure_proxy: float = 0.0
    jump_proxy: float = 0.0
</code>

Copy3. Структура фич

<code class="language-python">@dataclass
class MarketFeatures:
    mid_spot: float = 0.0
    spread_spot: float = 0.0
    qi: float = 0.0
    mli: float = 0.0
    cancel_pressure: float = 0.0
    trade_sign_int: float = 0.0
    book_thinning: float = 0.0
    bid_break_prob: float = 0.0

    rv_short: float = 0.0
    rv_medium: float = 0.0
    momentum_short: float = 0.0
    gap_risk: float = 0.0

    mid_fut: float = 0.0
    fut_momentum: float = 0.0
    fut_trade_sign: float = 0.0
    basis: float = 0.0
    basis_zscore: float = 0.0
    fut_liquidity: float = 0.0

    iv_atm: float = 0.0
    iv_skew: float = 0.0
    gamma_zone: float = 0.0
    option_jump_proxy: float = 0.0

    imoex_return: float = 0.0
    bank_sector_ret: float = 0.0
    usdrub_move: float = 0.0
    oil_move: float = 0.0
    market_stress: float = 0.0

    event_risk: float = 0.0
    forced_flow: float = 0.0

    sber_return: float = 0.0
    imoex_fut_return: float = 0.0
    bank_index_return: float = 0.0
    usdrub_cross_return: float = 0.0

    queue_bid_ahead: float = 0.0
    queue_ask_ahead: float = 0.0
    queue_fill_prob_bid: float = 0.0
    queue_fill_prob_ask: float = 0.0
</code>

Copy4. Параметры и лимиты

<code class="language-python">@dataclass
class Params:
    k_stress_beta: float = 0.5
    b_imoex: float = 0.8
    b_banks: float = 0.7
    b_fx: float = -0.2
    b_oil: float = 0.05
    b_momentum: float = 0.15
    b_fut_mom: float = 0.5
    b_basis: float = 0.2
    b_fut_flow: float = 0.25

    t_qi: float = 0.4
    t_mli: float = 0.3
    t_cancel: float = 0.2
    t_tradeflow: float = 0.35
    t_thinning: float = 0.25
    t_bidbreak: float = 0.5
    t_forced: float = 0.6
    t_futflow: float = 0.25

    j_gap: float = 0.4
    j_event: float = 0.7
    j_iv: float = 0.3
    j_opt: float = 0.4
    j_stress: float = 0.35

    inv_spot_penalty: float = 0.00002
    inv_fut_penalty: float = 0.00001
    s_ofi: float = 0.15
    s_qi: float = 0.10
    s_bidbreak: float = 0.20
    s_forced: float = 0.25
    s_inventory: float = 0.00003
    s_stress: float = 0.10
    s_futsignal: float = 0.12
    s_toxicity: float = 0.08

    min_spread: float = 0.0002
    max_spread: float = 0.01
    sp_rv_short: float = 0.8
    sp_rv_medium: float = 0.4
    sp_toxicity: float = 0.6
    sp_jump: float = 0.7
    sp_event: float = 0.5
    sp_basis: float = 0.2
    sp_hedge: float = 0.2
    sp_iv: float = 0.15
    sp_latency: float = 0.5

    base_size: float = 100000.0
    min_size: float = 1000.0
    max_size: float = 300000.0
</code>

Copy

<code class="language-python">@dataclass
class RiskLimits:
    event_hard_limit: float = 0.85
    forced_flow_limit: float = 0.75
    bid_break_limit: float = 0.75
    inventory_soft_limit: float = 300000.0
    inventory_hard_limit: float = 700000.0
    max_intraday_loss: float = -5_000_000.0
    max_feed_delay: float = 0.5
</code>

Copy

---

Как это собрать в аккуратный исследовательский проект

Я бы разложил так:

<code class="language-text">project/
├─ config.py          # Params, RiskLimits
├─ types.py           # dataclasses: OrderBook, Trade, Fill, MarketFeatures...
├─ data_api.py        # mock / replay data adapters
├─ execution_api.py   # simulated execution
├─ features.py        # feature engineering
├─ models.py          # fair value / toxicity / jump / regime / cross-impact
├─ strategy.py        # MarketMakerVTB
├─ backtest.py        # event loop over historical/replayed data
└─ notebooks/         # calibration / diagnostics / markout analysis
</code>

Copy

---

Что особенно важно методологически1. Этот каркас годится для:

• бэктеста,
• симуляции,
• stress‑testing,
• сравнения режимов,
• анализа markout,
• калибровки коэффициентов.

2. Его надо отделять от реального исполнения

Потому что в реальной системе отдельно живут:

• market data handler,
• order manager,
• risk engine,
• hedge engine,
• persistence,
• latency monitoring,
• failover.

Следующим этапом будет полноценный backtest.py для этого каркаса — с циклом по событиям, симуляцией fill и логированием PnL/markout