Add option for immediate execution with quotes inside the bid ask

# Conflicts:
#	nautilus_trader/backtest/engine.pyx
#	nautilus_trader/execution/engine.pyx

# Conflicts:
#	nautilus_trader/execution/engine.pyx

Author: faysou

examples/backtest/notebooks/databento_option_greeks.py

@@ -53,12 +53,12 @@
 from nautilus_trader.model.data import DataType
 from nautilus_trader.model.data import QuoteTick
 from nautilus_trader.model.enums import OrderSide
+from nautilus_trader.model.enums import PriceType
 from nautilus_trader.model.greeks_data import GreeksData
 from nautilus_trader.model.identifiers import InstrumentId
 from nautilus_trader.model.identifiers import Venue
 from nautilus_trader.model.identifiers import new_generic_spread_id
 from nautilus_trader.model.instruments import FuturesContract
-from nautilus_trader.model.objects import Price
 from nautilus_trader.model.objects import Quantity
 from nautilus_trader.model.tick_scheme import TieredTickScheme
 from nautilus_trader.model.tick_scheme import register_tick_scheme
@@ -156,6 +156,15 @@
     ],
 )
 
+
+def get_mid_price_rounded(tick, instrument, round_up=True, num_ticks=0):
+    mid_price = tick.extract_price(PriceType.MID)
+    if round_up:
+        return instrument.next_ask_price(mid_price.as_double(), num_ticks=num_ticks)
+    else:
+        return instrument.next_bid_price(mid_price.as_double(), num_ticks=num_ticks)
+
+
 # %% [markdown]
 # ## strategy
 
@@ -303,13 +312,31 @@ def on_quote_tick(self, tick):
         )
 
         # Submit spread order when we have spread quotes available
-        if tick.instrument_id == self.config.spread_id and not self.spread_order_submitted:
+        if (
+            tick.instrument_id == self.config.spread_id
+            and not self.spread_order_submitted
+            and tick.ts_init == 1715248980000000000
+        ):
             # Try submitting order immediately - the exchange should have processed the quote by now
-            self.submit_market_order(instrument_id=self.config.spread_id, quantity=5)
+            instrument = self.cache.instrument(tick.instrument_id)
+            mid_price_rounded = get_mid_price_rounded(
+                tick=tick,
+                instrument=instrument,
+                round_up=True,  # Round up for buy orders (more aggressive)
+            )
+            self.submit_limit_order(
+                instrument_id=self.config.spread_id,
+                price=mid_price_rounded,
+                quantity=5,
+            )
             self.spread_order_submitted = True
 
         if tick.instrument_id == self.config.spread_id2 and not self.spread_order_submitted2:
-            self.submit_market_order(instrument_id=self.config.spread_id2, quantity=5)
+            self.submit_limit_order(
+                instrument_id=self.config.spread_id2,
+                price=tick.ask_price,
+                quantity=5,
+            )
             self.spread_order_submitted2 = True
 
     def on_order_filled(self, event):
@@ -367,18 +394,18 @@ def submit_market_order(self, instrument_id, quantity):
             order_side=(OrderSide.BUY if quantity > 0 else OrderSide.SELL),
             quantity=Quantity.from_int(abs(quantity)),
         )
+        self.user_log(f"Submitting order: {order}")
         self.submit_order(order)
-        self.user_log(f"Order submitted: {order}")
 
     def submit_limit_order(self, instrument_id, price, quantity):
         order = self.order_factory.limit(
             instrument_id=instrument_id,
             order_side=(OrderSide.BUY if quantity > 0 else OrderSide.SELL),
             quantity=Quantity.from_int(abs(quantity)),
-            price=Price.from_str(f"{price:.2f}"),
+            price=price,
         )
+        self.user_log(f"Submitting order: {order}")
         self.submit_order(order)
-        self.user_log(f"Order submitted: {order}")
 
     def user_log(self, msg, color=LogColor.GREEN):
         self.log.warning(f"{msg}", color=color)
@@ -463,6 +490,7 @@ def on_stop(self):
     strategies=strategies,
     streaming=(streaming if not load_greeks else None),
     catalogs=catalogs,
+    # allow_immediate_quote_execution=True,
 )
 
 # BacktestRunConfig
@@ -498,12 +526,9 @@ def on_stop(self):
         *data,
     ]
 
-# Configure venue with enhanced SizeAwareFillModel for realistic option execution
-# This fill model provides different execution behavior based on order size:
-# - Small orders (<=10 contracts): Good liquidity at best prices
-# - Large orders: Experience price impact with partial fills at worse prices
+# Configure venue with enhanced BestPriceFillModel for being able to execute limit orders anywhere between a bid ask
 fill_model = ImportableFillModelConfig(
-    fill_model_path="nautilus_trader.backtest.models:SizeAwareFillModel",
+    fill_model_path="nautilus_trader.backtest.models:BestPriceFillModel",
     config_path="nautilus_trader.backtest.config:FillModelConfig",
     config={},
 )

nautilus_trader/backtest/config.py

@@ -334,6 +334,10 @@ class BacktestEngineConfig(NautilusKernelConfig, frozen=True):
         If logging should be bypassed.
     run_analysis : bool, default True
         If post backtest performance analysis should be run.
+    allow_immediate_quote_execution : bool, default False
+        Allows a strategy to receive quotes and pass orders before the data is
+        processed by the execution engine. Only relevant in backtesting.
+        Useful for backtests on discrete times, like every minute, hour or day.
 
     """
 
@@ -344,6 +348,7 @@ class BacktestEngineConfig(NautilusKernelConfig, frozen=True):
     risk_engine: RiskEngineConfig | None = RiskEngineConfig()
     exec_engine: ExecEngineConfig | None = ExecEngineConfig()
     run_analysis: bool = True
+    allow_immediate_quote_execution: bool = False
 
     def __post_init__(self):
         if isinstance(self.trader_id, str):