Source code for tradeexecutor.backtest.backtest_valuation
import datetime
from typing import Tuple
from tradeexecutor.backtest.backtest_pricing import BacktestPricing
from tradeexecutor.state.position import TradingPosition
from tradeexecutor.state.types import USDollarAmount
from tradeexecutor.state.identifier import TradingPairKind
from tradeexecutor.state.valuation import ValuationUpdate
from tradeexecutor.strategy.valuation import ValuationModel
[docs]class BacktestValuationModel(ValuationModel):
"""Re-value assets based on their on-chain backtest dataset price.
Each asset is valued at its market sell price estimation.
"""
[docs] def __init__(self, pricing_model: BacktestPricing):
assert pricing_model, "pricing_model missing"
self.pricing_model = pricing_model
[docs] def __call__(
self,
ts: datetime.datetime,
position: TradingPosition
) -> ValuationUpdate:
assert isinstance(ts, datetime.datetime)
assert ts.second == 0, f"Timestamp sanity check failed, does not have even seconds: {ts}"
pair = position.pair
if position.is_credit_supply():
# TODO: Assumes stable USD stablecoins
price = 1.0
elif position.is_long():
pricing_pair = pair.get_pricing_pair()
quantity = position.get_quantity()
trade_price = self.pricing_model.get_sell_price(ts, pricing_pair, quantity)
price = trade_price.price
else:
# TODO: Use position net asset pricing for leveraged positions
assert pair.kind == TradingPairKind.lending_protocol_short
quantity = -position.get_quantity()
trade_price = self.pricing_model.get_sell_price(ts, pair.underlying_spot_pair, quantity)
price = trade_price.price
old_price = position.last_token_price
old_value = position.get_value()
position.revalue_base_asset(ts, price)
new_value = position.get_value()
return ValuationUpdate(
position_id=position.position_id,
created_at=ts,
valued_at=ts,
new_price=price,
new_value=new_value,
old_value=old_value,
old_price=old_price,
)