import datetime
from decimal import Decimal
from typing import List, Optional
from tradeexecutor.backtest.simulated_wallet import SimulatedWallet
from tradeexecutor.ethereum.wallet import ReserveUpdateEvent
from tradeexecutor.state.portfolio import Portfolio
from tradeexecutor.state.identifier import AssetIdentifier
from tradeexecutor.state.state import State
from tradeexecutor.strategy.sync_model import SyncModel
from tradeexecutor.testing.dummy_wallet import apply_sync_events
[docs]class BacktestSyncer:
"""LEGACY backtest sync model.
Simulate deposit events to the backtest wallet."""
[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, portfolio: Portfolio, ts: datetime.datetime, supported_reserves: List[AssetIdentifier]) -> List[ReserveUpdateEvent]:
"""Process the backtest initial deposit.
The backtest wallet is credited once at the start.
"""
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
)
# Update wallet
self.wallet.update_balance(reserve_token.address, self.initial_deposit_amount)
# Update state
apply_sync_events(portfolio, [evt])
return [evt]
else:
return []
[docs]class BacktestSyncModel(SyncModel):
"""Backtest sync model.
Simulate deposit events to the backtest wallet."""
[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 sync_initial(self, state: State):
"""Set u[ initial sync details."""
deployment = state.sync.deployment
deployment.chain_id = None
deployment.address = None
deployment.block_number = None
deployment.tx_hash = None
deployment.block_mined_at = None
deployment.vault_token_name = None
deployment.vault_token_symbol = None
[docs] def sync_treasury(self,
strategy_cycle_ts: datetime.datetime,
state: State,
supported_reserves: Optional[List[AssetIdentifier]] = None
):
"""Apply the balance sync before each strategy cycle.
.. warning::
Old legacy code with wrong return signature compared to the parent class
"""
portfolio = state.portfolio
# TODO: Move this code eto sync_initial()
if not self.initial_deposit_processed_at:
self.initial_deposit_processed_at = strategy_cycle_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=strategy_cycle_ts
)
# Update wallet
self.wallet.update_balance(reserve_token.address, self.initial_deposit_amount)
# Update state
apply_sync_events(portfolio, [evt])
# Set synced flag
# TODO: fix - wrong event type
state.sync.treasury.last_updated_at = strategy_cycle_ts
state.sync.treasury.balance_update_refs = []
return []
else:
return []
[docs] def create_transaction_builder(self) -> None:
return None