algo_engine.apps

Application-layer tools including web dashboards, backtest UI, and simulated input.

Web Apps

class algo_engine.apps.backtest.WebApp(start_date: date, end_date: date, name: str = 'WebApp.Backtest', address: str = '0.0.0.0', port: int = 8080, **kwargs)

__init__(start_date: date, end_date: date, name: str = 'WebApp.Backtest', address: str = '0.0.0.0', port: int = 8080, **kwargs)

update(**kwargs)

register(ticker: str, **kwargs)

render_index()

render_dashboard(ticker: str)

serve(blocking: bool = True)

property url: str

algo_engine.apps.backtest.start_app(start_date: date, end_date: date, blocking: bool = True, **kwargs)

Bokeh Server

class algo_engine.apps.DocServer(theme: DocTheme = None, max_size: int = None, update_interval: float = 0.0, lock: lock = None, **kwargs)

__init__(theme: DocTheme = None, max_size: int = None, update_interval: float = 0.0, lock: lock = None, **kwargs)

abstractmethod update(timestamp: float, market_price: float, **kwargs)

abstractmethod update(timestamp: float, open_price: float, close_price: float, high_price: float, low_price: float, **kwargs)

abstractmethod update(market_data: MarketData, **kwargs)

abstractmethod layout(doc_id: int)

stream(doc_id: int = None)

patch(doc_id: int = None)

register_document(doc)

abstract property data: dict[str, list]

the data used to provide initial values for new bokeh.ColumnDataSource.

class algo_engine.apps.DocTheme

Candlestick Charts

class algo_engine.apps.backtest.CandleStick(ticker: str, start_date: date, end_date: date, interval: float = 60.0, x_axis: list[float] = None, theme: DocTheme = None, **kwargs)

class ActiveBarData

idx: int

ts_start: float

ts_end: float

open_price: float

close_price: float

high_price: float

low_price: float

volume: NotRequired[float]

__init__(ticker: str, start_date: date, end_date: date, interval: float = 60.0, x_axis: list[float] = None, theme: DocTheme = None, **kwargs)

ts_indices() → list[float]

generate integer indices from start date to end date, with given interval, in seconds

loc_indices(timestamp: float, start_idx: int = 0) → tuple[int, float]

update(**kwargs)

pipe(sequence: dict[str, list])

layout(doc_id: int)

to_csv(filename: str | Path)

property data: dict[str, list]

the data used to provide initial values for new bokeh.ColumnDataSource.

class algo_engine.apps.backtest.StickTheme(style: ColorStyle = None)

stick_padding = 0.1

range_padding = 0.01

class ColorStyle

up: str

down: str

ws_style = {'down': 'red', 'up': 'green'}

cn_style = {'down': 'green', 'up': 'red'}

__init__(style: ColorStyle = None)

stick_style(pct_change: float | int) → dict

Strategy Tester

class algo_engine.apps.Tester(start_date: date, end_date: date, dtype: list[str] = None, **kwargs)

__init__(start_date: date, end_date: date, dtype: list[str] = None, **kwargs)

register_ticker(ticker: str, **kwargs)

unregister_ticker(ticker: str, **kwargs)

buy(ticker: str, volume: float = None, limit_price: float = None)

sell(ticker: str, volume: float = None, limit_price: float = None)

abstractmethod load_data(ticker: str, market_date: date, dtype: Literal['TickData', 'TradeData', 'TransactionData', 'OrderBook']) → list[MarketData]

abstractmethod on_market_data(market_data: MarketData, **kwargs)

abstractmethod on_report(report: TradeReport, **kwargs)

abstractmethod on_order(order: TradeInstruction, **kwargs)

bod(market_date: date, **kwargs)

eod(market_date: date, **kwargs)

run(**kwargs)

class algo_engine.apps.StrategyTester(start_date: date, end_date: date, data_loader, strategy: StrategyEngine, **kwargs)

class StrategyEngine(position_tracker: PositionManagementService, **kwargs)

__init__(position_tracker: PositionManagementService, **kwargs)

add_handler(**kwargs)

add_handler_safe(**kwargs)

property algos

attach_strategy(strategy: object)

back_test(start_date: date, end_date: date, data_loader: Callable, **kwargs)

back_test_lite(start_date: date, end_date: date, data_loader: Callable, **kwargs)

bod(market_date: date, **kwargs)

cancel(ticker: str, side: TransactionSide = None, algo_id: str = None, order_id: str = None, **kwargs)

eod(market_date: date, **kwargs)

on_market_data(market_data: MarketData, **kwargs)

on_order(order: TradeInstruction, **kwargs)

on_report(report: TradeReport, **kwargs)

open_pos(ticker: str, volume: float, side: TransactionSide = None, limit_price: float = None, algo: str = None, **kwargs)

a method to open position :param ticker: the given ticker :param volume: the target open volume :param side: trade side :param limit_price: Optional limit :param algo: Optional the specified algo :param kwargs: other keyword used in algo :return:

register(event_engine=None, topic_set=None, auto_register: bool = True)

Register a virtual subclass of an ABC.

Returns the subclass, to allow usage as a class decorator.

remove_handler(**kwargs)

remove_handler_safe(**kwargs)

reset()

stop()

subscribe(ticker: str)

unregister(event_engine=None, topic_set=None, auto_unregister: bool = True)

unwind_pos(ticker: str, volume: float, side: TransactionSide = None, limit_price: float = None, algo: str = None, safe=True, **kwargs) → tuple[float, float]

unwind method provide a safe way to unwind position of given ticker.

Parameters:

• ticker – the given exposure

• volume – the target unwinding volume, should be a positive number

• side – the trade action side, e.g. if strategy wishes to sell (in order to unwind long position), then side = TransactionSide.Sell_to_Unwind

• limit_price – Optional, a limit price

• algo – Optional the algo to be used to execute unwinding action

• safe – True -> unwind volume should not exceed the exposed volume; False -> can flip position. Default is safe=True

• kwargs – other kwargs passing to algo.launch

Returns:

executed volume, remaining volume

__init__(start_date: date, end_date: date, data_loader, strategy: StrategyEngine, **kwargs)

register_ticker(ticker: str, **kwargs)

register()

Register a virtual subclass of an ABC.

Returns the subclass, to allow usage as a class decorator.

initialize_position_management()

load_data(ticker: str, market_date: date, dtype: Literal['TickData', 'TradeData', 'TransactionData', 'OrderBook']) → list[MarketData]

bod(market_date: date, **kwargs)

eod(market_date: date, **kwargs)

on_market_data(market_data: MarketData, **kwargs)

on_report(report: TradeReport, **kwargs)

on_order(order: TradeInstruction, **kwargs)

buy(ticker: str, volume: float = None, limit_price: float = None)

sell(ticker: str, volume: float = None, limit_price: float = None)

run(**kwargs)