OlympusTrader.strategy package¶

Submodules¶

OlympusTrader.strategy.base_strategy module¶

class OlympusTrader.strategy.base_strategy.BaseStrategy(BROKER: 'BaseBroker', RESOLUTION: 'ITimeFrame' = <factory>, MODE: 'IStrategyMode' = <IStrategyMode.BACKTEST: 'Backtest'>, STRATEGY_ID: 'UUID' = <factory>, NAME: 'str' = 'BaseStrategy', ACCOUNT: 'Optional[IAccount]' = <factory>, POSITIONS: 'dict[str, IPosition]' = <factory>, ORDERS: 'Optional[dict[str, IOrder]]' = <factory>, HISTORY: 'dict[str, pd.DataFrame]' = <factory>, INSIGHTS: 'dict[UUID, Insight]' = <factory>, UNIVERSE: 'dict[str, IAsset]' = <factory>, STREAMS: 'List[IMarketDataStream]' = <factory>, VARIABLES: 'AttributeDict' = <factory>, WITHUI: 'bool' = True, WITHSSM: 'bool' = True, SSM: 'Optional[SharedStrategyManager]' = None, tradeOnFeatureEvents: 'bool' = False, TOOLS: 'Optional[ITradingTools]' = None, VERBOSE: 'int' = 0, LOGGER: 'logging.Logger' = None, ALPHA_MODELS: 'List[BaseAlpha]' = <factory>, INSIGHT_EXECUTORS: 'dict[InsightState, deque[BaseExecutor]]' = <factory>, TaStrategy: 'ta.Study' = None, WARM_UP: 'int' = 0, execution_risk: 'float' = 0.01, minRewardRiskRatio: 'float' = 2.0, baseConfidence: 'float' = 0.1, shouldClosePartialFilledIfCancelled: 'bool' = True, insightRateLimit: 'int' = 1, BACKTESTING_CONFIG: 'IBacktestingConfig' = <factory>, BACKTESTING_RESULTS: 'dict[str, Portfolio]' = <factory>, METRICS: 'IStrategyMetrics' = <factory>, _watchdog_timeout: 'int' = 30, _teardown_timeout: 'int' = 30, _MAX_HISTORY_SIZE: 'int' = 1000)¶

Bases: ABC

ACCOUNT: IAccount | None¶: Account for the strategy

ALPHA_MODELS: List[BaseAlpha]¶: Alpha models to be used in the strategy

BACKTESTING_CONFIG: IBacktestingConfig¶: Backtesting configuration

BACKTESTING_RESULTS: dict[str, Portfolio]¶: Backtesting results

BROKER: BaseBroker¶: Broker for the strategy

HISTORY: dict[str, DataFrame]¶: History for the strategy

INSIGHTS: dict[UUID, Insight]¶: Insights for the strategy

INSIGHT_EXECUTORS: dict[InsightState, deque[BaseExecutor]]¶: Insight Executors Models

LOGGER: Logger = None¶

METRICS: IStrategyMetrics¶: Strategy metrics

MODE: IStrategyMode = 'Backtest'¶: Strategy mode

NAME: str = 'BaseStrategy'¶: Name of the strategy

ORDERS: dict[str, IOrder] | None¶: Orders for the strategy

POSITIONS: dict[str, IPosition]¶: Positions for the strategy

RESOLUTION: ITimeFrame¶: Resolution for the strategy

SSM: SharedStrategyManager | None = None¶: Shared Strategy Manager for the strategy functions

STRATEGY_ID: UUID¶: Unique ID for the strategy

STREAMS: List[IMarketDataStream]¶: Market Data Streams for the strategy

TOOLS: ITradingTools | None = None¶: Trading tools for the strategy

TaStrategy: Study = None¶: TA Strategy

UNIVERSE: dict[str, IAsset]¶: Universe of assets for the strategy

VARIABLES: AttributeDict¶: Variables for the strategy

VERBOSE: int = 0¶: Verbose level for the strategy

WARM_UP: int = 0¶: Warm up bars for the strategy to ensure the strategy has enough data to make decisions

WITHSSM: bool = True¶: Enable Shared Strategy Manager for the strategy functions

WITHUI: bool = True¶: Enable UI for the strategy

property account: IAccount | None¶: Returns the account of the strategy.

add_alpha(alpha: BaseAlpha)¶: Adds an alpha to the strategy.

add_alphas(alphas: List[BaseAlpha])¶: Adds a list of alphas to the strategy.

add_events(eventType: Literal['trade', 'quote', 'bar', 'news'] = 'bar', applyTA: bool = True, stored: bool = False, stored_path: str = 'data', time_frame: ITimeFrame | None = None, **kwargs)¶: Adds bar streams to the strategy.

add_executor(executor: BaseExecutor, state: InsightState = None)¶: Adds an executor to the strategy.

add_executors(executors: List[BaseExecutor], state: InsightState = None)¶: Adds a list of executors to the strategy.

add_insight(insight: Insight)¶: Adds an insight to the strategy.

add_ta(ta: List[dict])¶: Adds a technical analysis to the strategy.

property assets: dict[str, IAsset]¶: Returns the universe of the strategy.

baseConfidence: float = 0.1¶: Base Confidence level for the strategy

property broker: BaseBroker¶: Returns the broker used by the strategy.

async cancel_tracked_tasks() → None¶: Cancel tasks created with create_task().

close_position(symbol, qty=None, percent=None)¶: Cancels an order to the broker.

create_task(coro: Awaitable, name: str | None = None) → None¶

Create and track a background task on the running loop.

This helper is useful for launching tasks outside TaskGroup contexts. Tasks created here are stored in self._tasks and will be cancelled on teardown.

property current_datetime: <module 'datetime' from '/home/docs/.asdf/installs/python/3.12.10/lib/python3.12/datetime.py'>¶: Returns the current time of the strategy.

abstractmethod executeInsight(insight: Insight)¶

Called for each active insight in the strategy. it allows you to conrol the execution of the insight and manage the order. ### Example: ```python def executeInsight(self, insight: Insight):

match insight.state:

case InsightState.NEW:
pass

case InsightState.EXECUTED:
pass

case InsightState.FILLED:
pass

case InsightState.CLOSED:
pass

case InsightState.REJECTED:
pass

case InsightState.CANCELED:
pass

case _:
pass

```

execution_risk: float = 0.01¶: Execution risk per trade

abstractmethod generateInsights(symbol: str)¶: Called once per bar to generate insights.

property history: dict[str, DataFrame]¶: Returns the candle history of the strategy.

abstractmethod init(asset: IAsset)¶: Initialize the strategy. This method is called once before the start of the strategy on every asset in the universe.

insightRateLimit: int = 1¶: Insight rate limit

property insights: dict[UUID, Insight]¶: Returns the insights of the strategy.

property metrics: IStrategyMetrics¶: Returns the metrics of the strategy.

minRewardRiskRatio: float = 2.0¶: Minimum Reward to Risk Ratio required for the strategy

property mode: IStrategyMode¶: Returns the mode of the strategy.

abstractmethod on_bar(symbol: str, bar: dict)¶: Called once per bar. open, high, low, close, volume

property orders: dict[str, IOrder] | None¶: Returns the orders of the strategy.

property positions: dict[str, IPosition]¶: Returns the positions of the strategy.

property resolution: ITimeFrame¶: Returns the resolution of the strategy.

run()¶: Starts the strategy.

async run_teardown()¶: Clean up resources and save backtest results if applicable.

saveBacktestResults()¶: Save the backtest results.

shouldClosePartialFilledIfCancelled: bool = True¶: Insights that are partially filled and are cancelled should be closed if the insight is cancelled

abstractmethod start()¶: Start the strategy. This method is called once at the start of the strategy.

property state: dict¶: Returns the state of the strategy.

property streams: List[IMarketDataStream]¶: Returns the streams of the strategy.

submit_order(insight: Insight)¶: Submits an order to the broker.

abstractmethod teardown()¶: Called once at the end of the strategy.

property tools: ITradingTools¶: Returns the tools of the strategy.

tradeOnFeatureEvents: bool = False¶: Trade on feature events

abstractmethod universe() → set[str]¶: Used to generate the stock in play.

property warm_up: int¶: Returns the warm up period of the strategy.

OlympusTrader.strategy.interfaces module¶

class OlympusTrader.strategy.interfaces.IBacktestingConfig¶

Bases: TypedDict

preemptiveTA: bool | None = False¶

class OlympusTrader.strategy.interfaces.IMarketDataStream¶

Bases: TypedDict

TA: Study | None = None¶

applyTA: bool | None = False¶

asset_type: Literal['stock', 'crypto'] = 'crypto'¶

end: datetime | None = None¶

exchange: str¶

feature: str | None = None¶

start: datetime | None = None¶

stored: bool | None = False¶

stored_path: str | None = None¶

symbol: Required[str]¶

time_frame: Required[ITimeFrame]¶

type: Required[Literal['trade', 'quote', 'bar', 'news']] = 'bar'¶

class OlympusTrader.strategy.interfaces.IStrategyMetrics(start_date: datetime.datetime | None = None, end_date: datetime.datetime | None = None, starting_cash: float = 0.0, ending_cash: float = 0.0, total_open: int = 0, total_closed: int = 0, total_wins: int = 0, total_losses: int = 0, total_profit: float = 0.0, total_loss: float = 0.0, total_pnl: float = 0.0, win_rate: float = 0.0, avg_win: float = 0.0, avg_loss: float = 0.0)¶

Bases: object

avg_loss: float = 0.0¶: The average loss

avg_win: float = 0.0¶: The average win

end_date: datetime | None = None¶

ending_cash: float = 0.0¶: The ending cash

positionClosed(pnl: float)¶: Update the metrics when a position is closed

positionOpened()¶: Update the metrics when a position is opened

start_date: datetime | None = None¶: The start date

starting_cash: float = 0.0¶: The starting cash

total_closed: int = 0¶: The total number of closed trades

total_loss: float = 0.0¶: The total loss

total_losses: int = 0¶: The total number of losing trades

total_open: int = 0¶: The total number of open trades

total_pnl: float = 0.0¶: The total profit and loss

total_profit: float = 0.0¶: The total profit

total_wins: int = 0¶: The total number of winning trades

updateDerivedMetrics()¶: Update the derived metrics

updateEnd(end_date: datetime, ending_cash: float)¶: Update the end date

updateStart(start_date: datetime, starting_cash: float)¶: Update the start date and starting cash

win_rate: float = 0.0¶: The total win rate

class OlympusTrader.strategy.interfaces.IStrategyMode(*values)¶

Bases: Enum

BACKTEST = 'Backtest'¶

LIVE = 'Live'¶

OlympusTrader.strategy.sharedmemory module¶

class OlympusTrader.strategy.sharedmemory.SharedStrategyManager(address=None, authkey=None, serializer='pickle', ctx=None, *, shutdown_timeout=1.0)¶

Bases: BaseManager

get_account(*args, **kwds)¶

get_assets(*args, **kwds)¶

get_history(*args, **kwds)¶

get_insights(*args, **kwds)¶

get_metrics(*args, **kwds)¶

get_mode(*args, **kwds)¶

get_positions(*args, **kwds)¶

get_strategy(*args, **kwds)¶

get_time(*args, **kwds)¶

OlympusTrader.strategy.strategy module¶

class OlympusTrader.strategy.strategy.Strategy(broker: ~OlympusTrader.broker.base_broker.BaseBroker, variables: ~OlympusTrader.utils.types.AttributeDict = {}, resolution: ~OlympusTrader.utils.timeframe.ITimeFrame = <OlympusTrader.utils.timeframe.ITimeFrame object>, verbose: int = 0, ui: bool = True, ssm: bool = True, mode: ~OlympusTrader.strategy.interfaces.IStrategyMode = IStrategyMode.LIVE, tradeOnFeatureEvents: bool = False, **kwargs)¶

Bases: BaseStrategy

executeInsight(insight: Insight)¶