algo_engine.engine

The engine layer provides event-driven dispatch, market data management, algorithm execution, and trade order routing.

Event Engine

algo_engine.engine.EVENT_ENGINE = <EventEngineEx idle>(capacity=4095, timers=[])

algo_engine.engine.TOPIC

alias of <algo_engine.engine.c_event_engine.TopicSet object>

Market Data Service

class algo_engine.engine.MarketDataService

add_monitor(self, monitor) → void

clear(self) → void

get_market_price(self, str ticker) → double

market_date

market_price

market_time

monitor

monitor_manager

n_subscribed

on_internal_data(self, InternalData internal_data) → void

on_market_data(self, MarketData market_data) → void

pop_monitor(self, monitor=None, monitor_id=None, str monitor_name=None) → void

profile

set_manager(self, MonitorManager manager) → void

subscriptions

timestamp

algo_engine.engine.MDS = <algo_engine.engine.c_market_engine.MarketDataService object>

class algo_engine.engine.MarketDataMonitor(name: str, monitor_id: Any = None)

__init__(self, str name: str, monitor_id: Any = None)

abstractmethod clear(self) → None

MarketDataMonitor.is_ready -> bool

MarketDataMonitor.value -> dict[str, float] | float

class algo_engine.engine.MonitorManager

add_monitor(self, monitor) → void

clear(self) → void

clear_monitors(self) → void

feed_monitor(self, monitor_id, MarketData market_data) → void

get_values(self) → dict

monitor

on_market_data(self, MarketData market_data) → void

pop_monitor(self, monitor_id) → void

start(self) → void

stop(self) → void

values

Algo Engine

class algo_engine.engine.AlgoTemplate(dma: DirectMarketAccess, ticker: str, target_volume: float, side: TransactionSide, **kwargs)

Status

alias of AlgoStatus

__init__(dma: DirectMarketAccess, ticker: str, target_volume: float, side: TransactionSide, **kwargs)

Template for trading algorithm an abstract class to create a trading algorithm

Parameters:

• dma – direct market access

• ticker – the given symbol of the underlying to trade

• target_volume – the given volume to trade

• side – the given TransactionSide

Keyword Arguments:

• algo_engine – the algo_engine instance, default is ALGO_ENGINE

• logger – the logger instance, default is LOGGER

• algo_id – the id of the algo, default is uuid4()

on_sync_progress(progress: float, **kwargs)

on_market_data(market_data: MarketData, **kwargs)

on_filled(report: TradeReport, **kwargs)

on_canceled(order_id: str | bytes | int | UUID = None, **kwargs)

on_rejected(order: TradeInstruction, **kwargs)

recover()

to_json(fmt='str') → str | dict

from_json(json_str: str | dict)

abstractmethod work()

abstractmethod launch(**kwargs) → list[TradeInstruction]

launch is a method to initiate the algo and launching orders. this method will set the algo is_active = true this method will set a new algo state, usually idle -> working launch method is designed to be called by strategy / position management service.

Parameters:

kwargs – other keywords needed to launch an algo

Returns:

a list of working orders. Noted, that not all working order is returned by this method, for example, TWAP algo will init a sequence of order and return later.

abstractmethod cancel(**kwargs)

cancel is a method to cancel / stop ALL working orders this method will set the algo is_active = false this method may set a new algo state, usually working -> stopping launch method is designed to be called by strategy / position management service.

Parameters:

kwargs – other keywords needed to cancel an algo

Returns:

None

property trades: dict[str, TradeReport]

property average_price: float

property exposure_volume: float

<WITH SIGN> net exposed VOLUME indicating the exposure of the pos :return: float

property working_volume: float

<WITHOUT SIGN> net working VOLUME indicating the working status of the pos :return: float

property filled_volume: float

<WITHOUT SIGN> filled VOLUME :return: float

property filled_notional: float

<POSSIBLY WITH SIGN> total filled Notional :return: float

property fee: float

<POSSIBLY WITH SIGN> total transaction fee :return: float

property cash_flow: float

<WITH SIGN> total cash flow :return: float

property multiplier: float

property filled_progress

property placed_progress

property target_progress

property market_price

property market_time: datetime

property timestamp: float

property start_time: datetime | None

property finish_time: datetime | None

class algo_engine.engine.algo_engine.AlgoEngine(*args, **kwargs)

__init__(mds=None, registry=None)

get_algo(name: str)

calculate_limit(algo: AlgoTemplate, limit_price: float = None, limit_adjust_factor: float = None, limit_adjust_level: float = None, mode: str = 'loose') → float | None

Calculate limit price

Parameters:

• algo – given algo

• limit_price – absolute limit_price

• limit_adjust_factor – limit_price = market_price * (1 + factor) for long order else limit price = market_price * (1 - factor) for short order

• limit_adjust_level – for long order, limit price = bid[lvl] if lvl > 0 else ask[lvl] for lvl < 0.

• mode – “strict” to select strictest limit price or “loose” to select loosest one. Default is None, which is “strict”.

Returns:

the calculated limit price, if there is any

from_json(json_str, dma) → AlgoTemplate

class algo_engine.engine.algo_engine.AlgoRegistry(*args, **kwargs)

registry for trade algos

to add a new algo, add name to __init__ method, add handler to .cast() method

DO NOT add any other value to __init__.

__init__()

add_algo(name: str, *alias, handler: type[AlgoTemplate])

cast(value: str)

property reversed_registry: dict[str, str]

to_algo(name: str, algo_engine: AlgoEngine = None)

algo_engine.engine.ALGO_ENGINE = <algo_engine.engine.algo_engine.AlgoEngine object>

algo_engine.engine.ALGO_REGISTRY = <algo_engine.engine.algo_engine.AlgoRegistry object>

registry for trade algos

to add a new algo, add name to __init__ method, add handler to .cast() method

DO NOT add any other value to __init__.

Trade Engine

class algo_engine.engine.DirectMarketAccess(mds: MarketDataService, risk_profile: RiskProfile, cool_down: float = None)

Direct Market Access

send launch/cancel order direct to market(exchange)

also contains an order buff designed to process order and control risk

2 ways to implement this api - override the abstractmethod _launch_order_handler, _cancel_order_handler, _reject_order_handler to api directly - or use event engine

__init__(mds: MarketDataService, risk_profile: RiskProfile, cool_down: float = None)

launch_order(order: TradeInstruction, **kwargs)

cancel_order(order: TradeInstruction, **kwargs)

start()

shut_down()

property timestamp

property market_price

property market_time

class algo_engine.engine.Balance(*args, **kwargs)

Balance handles mapping of PositionTracker <-> Strategy

__init__(inventory: Inventory = None)

add(map_id: str = None, strategy=None, position_tracker: PositionManagementService = None)

pop(map_id: str)

get(**kwargs) → PositionManagementService | None

get_strategy(strategy_name: str = None, strategy_id=None)

get_tracker(strategy_name: str = None, strategy_id=None) → PositionManagementService | None

on_update(market_time=None)

on_order(order: TradeInstruction, **kwargs)

on_report(report: TradeReport, **kwargs)

reset()

to_json(fmt='str') → str | dict

from_json(json_str: str | dict)

dump(file_path: str | Path)

dump_trades(file_path: Path | str = None, ts_from: float = None, ts_to: float = None) → dict

export all trade monitored by position manager

Parameters:

• file_path – Optional, the exported path, without it, the dict will not be dumped

• ts_from – timestamp from

• ts_to – timestamp to

Returns:

a dict containing all the trades

dump_trades_all(file_path: Path | str = None, ts_from: float = None, ts_to: float = None) → list

export all the trades received by Balance module, even if there is no strategy corresponding to it.

Parameters:

• file_path – Optional, the exported path, without it, the dict will not be dumped

• ts_from – timestamp from

• ts_to – timestamp to

Returns:

a list containing all the trades info

load(file_path: str | Path)

property tracker_mapping: dict[str, str]

property reversed_tracker_mapping: dict[str, str]

property strategy_mapping: dict[str, int]

property reversed_strategy_mapping: dict[int, str]

property working_volume_summed: dict[str, float]

property exposure_volume: dict[str, float]

property working_volume: dict[str, dict[str, float]]

exposure_notional(mds) → dict[str, float]

working_notional(mds) → dict[str, float]

property orders: dict[str, TradeInstruction]

property working_order: dict[str, TradeInstruction]

property trades_today

property trades_session: dict[str, TradeReport]

property trades: dict[str, TradeReport]

property info: DataFrame

class algo_engine.engine.PositionManagementService(dma: DirectMarketAccess, algo_engine=None, default_algo: str = None, no_cache: bool = False, **kwargs)

Position Module controls the position of a single strategy,

The tracker provides basic tracing of PnL, exposure, holding time and interface with risk monitor module The Strategy should interface with Position module, not the algo

a range of easy method is provided to facilitate development

__init__(dma: DirectMarketAccess, algo_engine=None, default_algo: str = None, no_cache: bool = False, **kwargs)

on_market_data(market_data: MarketData)

on_filled(report: TradeReport, **kwargs)

on_canceled(order_id: str, **kwargs)

on_rejected(order: TradeInstruction, **kwargs)

on_algo_done(algo: AlgoTemplate)

on_algo_error(algo: AlgoTemplate)

open(ticker: str, target_volume: float, trade_side: TransactionSide, algo: str = None, **kwargs)

unwind_ticker(ticker: str, **kwargs)

add_exposure(ticker: str, volume: float, notional: float, side: TransactionSide, timestamp: float)

this is a method to add dummy algo and fills it.

the method provides an easy way to amend exposure

unwind_all(**kwargs)

cancel_all(**kwargs)

to_json(fmt='str') → str | dict

clear_cache()

clear()

pnl() → dict[str, float]

property notional: dict[str, float]

property working_volume: dict[str, dict[str, float]]

a dictionary indicating current working volume of all orders

{‘Long’: +float, ‘Short’: +float}

Returns:

a dict with non-negative numbers

property exposure_volume: dict[str, float]

a dictionary indicating current net exposed volume of all orders

Returns:

a dict with float numbers (positive and negatives)

property working_volume_net: dict[str, float]

a dictionary indicating current working volume of all orders

Returns:

a dict with summed working volume for each ticker numbers, with positive value as net-long and negative value as net-short

property market_price

property market_time

property orders: dict[str, TradeInstruction]

property working_order: dict[str, TradeInstruction]

property trades: dict[str, TradeReport]

property order_mapping: dict[str, dict[str, TradeInstruction]]

property reversed_order_mapping: dict[str, AlgoTemplate]

class algo_engine.engine.RiskProfile(*args, **kwargs)

exception Risk(risk_type: str, code: int, msg: str, *args, **kwargs)

__init__(risk_type: str, code: int, msg: str, *args, **kwargs)

__init__(mds: MarketDataService, balance: Balance, **kwargs)

set_rule(key: str, value: float, ticker: str = None)

get(ticker: str) → dict[str, float | dict[str, float]]

check(order: TradeInstruction)

check_order(ticker: str, volume: float, side: TransactionSide)

check_basket(*order: TradeInstruction)

clear()

to_json(fmt='str') → str | dict

from_json(json_str: str | dict)

dump(file_path: str | Path)

load(file_path: str | Path)

property market_time

property info: DataFrame

Singleton

class algo_engine.engine.Singleton