Overall Statistics
Total Orders
3
Average Win
0%
Average Loss
0%
Compounding Annual Return
44.434%
Drawdown
3.600%
Expectancy
0
Start Equity
100000.00
End Equity
103275.77
Net Profit
3.276%
Sharpe Ratio
2.109
Sortino Ratio
3.362
Probabilistic Sharpe Ratio
67.045%
Loss Rate
0%
Win Rate
0%
Profit-Loss Ratio
0
Alpha
0
Beta
0
Annual Standard Deviation
0.116
Annual Variance
0.013
Information Ratio
2.584
Tracking Error
0.116
Treynor Ratio
0
Total Fees
$2.25
Estimated Strategy Capacity
$24000000.00
Lowest Capacity Asset
FESX YGT6HGVF2U1X
Portfolio Turnover
3.13%
# region imports
from AlgorithmImports import *
# endregion

class EurexEuroStoxx50Test(QCAlgorithm):

    def __init__(self):        
        self._traded = False
        self._liquidated = False
        self._future: Future = None
        self._index: Index = None

        self._limit_order_ticket_to_cancel: OrderTicket = None
        self._limit_order_ticket_to_update: OrderTicket = None

    def initialize(self):
        self.set_start_date(2024, 1, 1)
        self.set_end_date(2024, 2, 1)
        
        self._future = self.add_future(Futures.Indices.EURO_STOXX_50, Resolution.MINUTE,
                data_normalization_mode=DataNormalizationMode.BACKWARDS_RATIO,
                data_mapping_mode=DataMappingMode.FIRST_DAY_MONTH,
                contract_depth_offset=0)
        self._future.set_filter(0, 180)

        self._index = self.add_index("SX5E", Resolution.MINUTE, market=Market.EUREX)

        # self.set_warm_up(30, Resolution.DAILY)

        self.set_benchmark(lambda x: 0)

    def is_symbol_of_interest(self, symbol: Symbol) -> bool:
        return symbol == self._future.symbol or \
            (symbol.has_canonical() and symbol.canonical == self._future.symbol) or \
                symbol == self._index.symbol
        
    def on_warmup_finished(self):
        # self.log(f"[{self.time}] :: Mapped contract: {self._future.mapped.value}")

        # securities = [security for security in self.securities.values() if self.is_symbol_of_interest(security.symbol)]
        # prices_str = ", ".join([f"{security.Symbol.Value}: {security.Price}" for security in securities])
        # self.log(f"[{self.time}] :: Security Prices: {prices_str}")

        # Place trades
        # self.place_trades()
        pass

    def place_trades(self):
        self.buy(self._future.mapped, 1)
        
        # Limit orders that won't fill
        self._limit_order_ticket_to_cancel = self.limit_order(self._future.mapped, 1, 3000, tag="To cancel")
        self._limit_order_ticket_to_update = self.limit_order(self._future.mapped, 1, 3000, tag="To update")

    def on_order_event(self, order_event: OrderEvent):
        if order_event.ticket.order_type == OrderType.LIMIT and order_event.ticket.tag == "To cancel":
            order_event.ticket.cancel()

        if order_event.ticket.order_type == OrderType.LIMIT and order_event.ticket.tag == "To update":
            order_event.ticket.update_limit_price(
                self._future.price, 
                f"{order_event.ticket.tag}: Updated limit price from {order_event.limit_price} to {self._future.price}")

    def on_data(self, slice: Slice):
        if self.is_warming_up: 
            return
        
        data_str = "\n".join([f"[{SecurityType(data.symbol.security_type)}] [{data.symbol.value}] {data}" for data in slice.all_data])
        self.log(f"[{self.time}] :: Data:\n{data_str}")              

        if not self._traded and self._future.mapped is not None:
            self.place_trades()
            self._traded = True