Overall Statistics
Total Orders
0
Average Win
0%
Average Loss
0%
Compounding Annual Return
0%
Drawdown
0%
Expectancy
0
Start Equity
100000
End Equity
100000
Net Profit
0%
Sharpe Ratio
0
Sortino Ratio
0
Probabilistic Sharpe Ratio
0%
Loss Rate
0%
Win Rate
0%
Profit-Loss Ratio
0
Alpha
0
Beta
0
Annual Standard Deviation
0
Annual Variance
0
Information Ratio
-1.3
Tracking Error
0.116
Treynor Ratio
0
Total Fees
$0.00
Estimated Strategy Capacity
$0
Lowest Capacity Asset
Portfolio Turnover
0%
Drawdown Recovery
0
 
# region imports
from AlgorithmImports import *
# endregion

class ESHistoryAndStreamingComparison(QCAlgorithm):

    def initialize(self):
        # Run across the target day and one day after so history can query a completed session
        self.set_start_date(2025, 9, 24)
        self.set_end_date(self.start_date + timedelta(2))
        self.target_start = self.start_date
        self.target_end   = self.start_date+timedelta(1)


        # Continuous ES subscription
        self.future = self.add_future(
            "NQ",
            resolution=Resolution.MINUTE,
            fill_forward=False,
            extended_market_hours=True,
            data_mapping_mode=DataMappingMode.LAST_TRADING_DAY,
            data_normalization_mode=DataNormalizationMode.RAW,
            contract_depth_offset=0
        )

    def on_end_of_algorithm(self):
        self.log("========== HISTORY COMPARISON =========="
            f"Algorithm time: {self.time}")

        def get_history(symbol, fill_forward):
            df = self.history(
            symbol,
            self.target_start,
            self.target_end,
            Resolution.MINUTE,
            fill_forward=fill_forward,
            extended_market_hours=True,
            data_mapping_mode=DataMappingMode.LAST_TRADING_DAY,
            data_normalization_mode=DataNormalizationMode.RAW,
            contract_depth_offset=0)
            df = df[df['volume']!=0]
            self.log(f"HIST {symbol.value} FF={fill_forward}. Shape={df.shape}")
            return df
        
        # Continuous history
        hist_cont_noff = get_history(self.future.symbol, False)
        hist_cont_ff = get_history(self.future.symbol, True)

        # Mapped history
        mapped = self.future.mapped
        if mapped:
            hist_map_noff = get_history(mapped, False)
            hist_map_ff = get_history(mapped, True)