Source code for tradeexecutor.backtest.legacy_backtest_sync

"""Old backtest wallet top up code."""

import datetime
from _decimal import Decimal
from typing import List

from tradeexecutor.backtest.simulated_wallet import SimulatedWallet
from tradeexecutor.ethereum.wallet import ReserveUpdateEvent
from tradeexecutor.state.identifier import AssetIdentifier
from tradeexecutor.state.state import State
from tradeexecutor.testing.dummy_wallet import apply_sync_events

[docs]class BacktestSyncer: """LEGACY backtest sync model. Simulate deposit events to the backtest wallet. .. warning:: Does not correctly fire any balance update events. Can be used to backtest with a fixed initial amount only. """
[docs] def __init__(self, wallet: SimulatedWallet, initial_deposit_amount: Decimal): assert isinstance(initial_deposit_amount, Decimal) self.wallet = wallet self.initial_deposit_amount = initial_deposit_amount self.initial_deposit_processed_at = None
[docs] def __call__(self, state: State, ts: datetime.datetime, supported_reserves: List[AssetIdentifier]) -> List[ReserveUpdateEvent]: """Process the backtest initial deposit. The backtest wallet is credited once at the start. """ assert isinstance(state, State) portfolio = state.portfolio if not self.initial_deposit_processed_at: self.initial_deposit_processed_at = ts assert len(supported_reserves) == 1 reserve_token = supported_reserves[0] # Generate a deposit event evt = ReserveUpdateEvent( asset=reserve_token, past_balance=Decimal(0), new_balance=self.initial_deposit_amount, updated_at=ts, mined_at=ts, ) # Update wallet self.wallet.update_token_info(reserve_token) self.wallet.update_balance(reserve_token.address, self.initial_deposit_amount) # Update state apply_sync_events(state, [evt]) return [evt] else: return []