diff --git a/tests/execution/backtest/test_bid_ask_fills.py b/tests/execution/backtest/test_bid_ask_fills.py
new file mode 100644
index 0000000..70a294c
--- /dev/null
+++ b/tests/execution/backtest/test_bid_ask_fills.py
@@ -0,0 +1,241 @@
+"""Tests for bid/ask-aware fill pricing and transaction cost overlays."""
+
+from __future__ import annotations
+
+import logging
+
+import pytest
+
+from tradedesk.execution.backtest.client import BacktestClient, TransactionCosts
+from tradedesk.execution.backtest.streamer import CandleSeries
+from tradedesk.types import Candle
+
+
+def _make_candle(ts: str, close: float) -> Candle:
+    return Candle(timestamp=ts, open=close, high=close, low=close, close=close)
+
+
+def _client_with_bid_ask(
+    bid_close: float,
+    ask_close: float,
+    ts: str = "2025-01-01T00:00:00Z",
+) -> BacktestClient:
+    """Create a BacktestClient with both bid and ask prices set."""
+    candle = _make_candle(ts, bid_close)
+    ask_candle = _make_candle(ts, ask_close)
+
+    client = BacktestClient.from_history({("INST", "1MIN"): [candle]})
+    client.set_ask_series(
+        [CandleSeries(instrument="INST", period="1MIN", candles=[ask_candle])]
+    )
+    client._mark_price["INST"] = bid_close
+    client._ask_price["INST"] = ask_close
+    client._current_timestamp = ts
+    return client
+
+
+@pytest.mark.asyncio
+async def test_buy_fills_at_ask() -> None:
+    """BUY orders must fill at the ask price, not the bid."""
+    client = _client_with_bid_ask(bid_close=100.0, ask_close=100.5)
+    await client.start()
+
+    result = await client.place_market_order("INST", "BUY", size=1.0)
+
+    assert result["price"] == 100.5
+    assert client.trades[0].price == 100.5
+    assert client.trades[0].raw_price == pytest.approx(100.25)  # mid = (100 + 100.5) / 2
+    assert client.trades[0].spread_cost == pytest.approx(0.25)  # half-spread
+
+
+@pytest.mark.asyncio
+async def test_sell_fills_at_bid() -> None:
+    """SELL orders must fill at the bid price."""
+    client = _client_with_bid_ask(bid_close=100.0, ask_close=100.5)
+    await client.start()
+
+    result = await client.place_market_order("INST", "SELL", size=1.0)
+
+    assert result["price"] == 100.0
+    assert client.trades[0].price == 100.0
+    assert client.trades[0].raw_price == pytest.approx(100.25)
+    assert client.trades[0].spread_cost == pytest.approx(0.25)
+
+
+@pytest.mark.asyncio
+async def test_round_trip_pnl_includes_spread() -> None:
+    """A long round trip PnL is (bid_exit - ask_entry) * size."""
+    bid_entry = 100.0
+    ask_entry = 100.5
+    bid_exit = 102.0
+    ask_exit = 102.5
+
+    client = _client_with_bid_ask(bid_close=bid_entry, ask_close=ask_entry)
+    await client.start()
+
+    await client.place_market_order("INST", "BUY", size=2.0)  # fills at 100.5
+
+    # Update prices for exit
+    client._mark_price["INST"] = bid_exit
+    client._ask_price["INST"] = ask_exit
+    client._current_timestamp = "2025-01-01T01:00:00Z"
+
+    await client.place_market_order("INST", "SELL", size=2.0)  # fills at 102.0
+
+    # PnL = (102.0 - 100.5) * 2 = 3.0
+    assert client.realised_pnl == pytest.approx(3.0)
+    assert client.positions == {}
+
+
+@pytest.mark.asyncio
+async def test_no_ask_series_falls_back_to_bid() -> None:
+    """When no ask series is set, all fills use bid price (no spread cost)."""
+    candle = _make_candle("2025-01-01T00:00:00Z", 100.0)
+    client = BacktestClient.from_history({("INST", "1MIN"): [candle]})
+    client._mark_price["INST"] = 100.0
+    client._current_timestamp = "2025-01-01T00:00:00Z"
+    await client.start()
+
+    result = await client.place_market_order("INST", "BUY", size=1.0)
+
+    assert result["price"] == 100.0
+    assert client.trades[0].spread_cost == 0.0
+    assert client.trades[0].raw_price == 100.0  # falls back to bid price when no ask data
+
+
+@pytest.mark.asyncio
+async def test_missing_ask_price_at_fill_logs_warning(caplog: pytest.LogCaptureFixture) -> None:
+    """When ask series is registered but ask price not yet set, a warning is emitted."""
+    candle = _make_candle("2025-01-01T00:00:00Z", 100.0)
+    ask_candle = _make_candle("2025-01-02T00:00:00Z", 100.5)  # different timestamp
+
+    client = BacktestClient.from_history({("INST", "1MIN"): [candle]})
+    client.set_ask_series(
+        [CandleSeries(instrument="INST", period="1MIN", candles=[ask_candle])]
+    )
+    client._mark_price["INST"] = 100.0
+    client._current_timestamp = "2025-01-01T00:00:00Z"
+    # _ask_price["INST"] is NOT set — simulates a gap in ask data
+    await client.start()
+
+    with caplog.at_level(logging.WARNING, logger="tradedesk.execution.backtest.client"):
+        result = await client.place_market_order("INST", "BUY", size=1.0)
+
+    assert result["price"] == 100.0  # fell back to bid
+    assert any("Missing ask price" in r.message for r in caplog.records)
+
+
+@pytest.mark.asyncio
+async def test_slippage_points_applied_adversely() -> None:
+    """Fixed slippage in points worsens the fill price for both BUY and SELL."""
+    tc = TransactionCosts(slippage_points=0.5)
+    client = _client_with_bid_ask(bid_close=100.0, ask_close=100.5)
+    client.set_transaction_costs(tc)
+    await client.start()
+
+    # BUY: fills at ask + slippage = 100.5 + 0.5 = 101.0
+    result_buy = await client.place_market_order("INST", "BUY", size=1.0)
+    assert result_buy["price"] == pytest.approx(101.0)
+    assert client.trades[0].slippage_cost == pytest.approx(0.5)
+
+    # SELL: fills at bid - slippage = 100.0 - 0.5 = 99.5
+    client._mark_price["INST"] = 100.0
+    client._ask_price["INST"] = 100.5
+    result_sell = await client.place_market_order("INST", "SELL", size=1.0)
+    assert result_sell["price"] == pytest.approx(99.5)
+    assert client.trades[1].slippage_cost == pytest.approx(0.5)
+
+
+@pytest.mark.asyncio
+async def test_slippage_bps_applied() -> None:
+    """Proportional slippage in bps is applied to fill price."""
+    tc = TransactionCosts(slippage_bps=10)  # 10 bps = 0.1%
+    client = _client_with_bid_ask(bid_close=100.0, ask_close=100.5)
+    client.set_transaction_costs(tc)
+    await client.start()
+
+    # BUY: slippage = 100.5 * 10 / 10000 = 0.1005; fill = 100.5 + 0.1005 = 100.6005
+    result = await client.place_market_order("INST", "BUY", size=1.0)
+    assert result["price"] == pytest.approx(100.5 + 100.5 * 10 / 10_000)
+
+
+@pytest.mark.asyncio
+async def test_commission_per_fill_deducted_from_realised_pnl() -> None:
+    """Commission per fill is deducted from realised PnL on every order."""
+    tc = TransactionCosts(commission_per_fill=1.5)
+    client = _client_with_bid_ask(bid_close=100.0, ask_close=100.0)
+    client.set_transaction_costs(tc)
+    await client.start()
+
+    await client.place_market_order("INST", "BUY", size=1.0)
+    # After entry: -1.5 commission
+    assert client.realised_pnl == pytest.approx(-1.5)
+    assert client.trades[0].commission_cost == pytest.approx(1.5)
+
+    client._mark_price["INST"] = 110.0
+    client._ask_price["INST"] = 110.0
+    client._current_timestamp = "2025-01-01T01:00:00Z"
+
+    await client.place_market_order("INST", "SELL", size=1.0)
+    # After exit: gross PnL = 10.0, commissions = 2 * 1.5 = 3.0 → net = 7.0
+    assert client.realised_pnl == pytest.approx(7.0)
+
+
+@pytest.mark.asyncio
+async def test_commission_per_round_trip_deducted_at_close() -> None:
+    """Commission per round trip is charged once at position close."""
+    tc = TransactionCosts(commission_per_round_trip=5.0)
+    client = _client_with_bid_ask(bid_close=100.0, ask_close=100.0)
+    client.set_transaction_costs(tc)
+    await client.start()
+
+    await client.place_market_order("INST", "BUY", size=1.0)
+    assert client.realised_pnl == pytest.approx(0.0)  # no commission at entry
+
+    client._mark_price["INST"] = 110.0
+    client._ask_price["INST"] = 110.0
+    client._current_timestamp = "2025-01-01T01:00:00Z"
+
+    await client.place_market_order("INST", "SELL", size=1.0)
+    # gross = 10.0, round-trip commission = 5.0 → net = 5.0
+    assert client.realised_pnl == pytest.approx(5.0)
+
+
+@pytest.mark.asyncio
+async def test_position_closed_event_carries_cost_fields() -> None:
+    """PositionClosedEvent includes full cost decomposition from the client."""
+    from tradedesk.events import get_dispatcher
+    from tradedesk.recording import PositionClosedEvent
+
+    dispatcher = get_dispatcher()
+    closed_events: list[PositionClosedEvent] = []
+
+    async def capture(ev: PositionClosedEvent) -> None:
+        closed_events.append(ev)
+
+    dispatcher.subscribe(PositionClosedEvent, capture)
+
+    tc = TransactionCosts(slippage_points=0.1, commission_per_fill=0.5)
+    client = _client_with_bid_ask(bid_close=100.0, ask_close=100.5)
+    client.set_transaction_costs(tc)
+    await client.start()
+
+    await client.place_market_order("INST", "BUY", size=1.0)
+
+    client._mark_price["INST"] = 102.0
+    client._ask_price["INST"] = 102.6
+    client._current_timestamp = "2025-01-01T01:00:00Z"
+    await client.place_market_order("INST", "SELL", size=1.0)
+
+    assert len(closed_events) == 1
+    ev = closed_events[0]
+    assert ev.raw_entry_price == pytest.approx(100.25)  # mid
+    assert ev.raw_exit_price == pytest.approx(102.3)   # mid
+    assert ev.entry_spread_cost == pytest.approx(0.25)
+    assert ev.exit_spread_cost == pytest.approx(0.3)
+    assert ev.entry_slippage_cost == pytest.approx(0.1)
+    assert ev.exit_slippage_cost == pytest.approx(0.1)
+    assert ev.entry_commission_cost == pytest.approx(0.5)
+    assert ev.exit_commission_cost == pytest.approx(0.5)
+
+    dispatcher.unsubscribe(PositionClosedEvent, capture)
diff --git a/tradedesk/execution/backtest/__init__.py b/tradedesk/execution/backtest/__init__.py
index 8fc572f..1e05524 100644
--- a/tradedesk/execution/backtest/__init__.py
+++ b/tradedesk/execution/backtest/__init__.py
@@ -1,6 +1,6 @@
 """Backtesting provider implementation."""
 
-from .client import BacktestClient
+from .client import BacktestClient, TransactionCosts
 from .dukascopy import iter_dukascopy_candles, read_dukascopy_candles
 from .runner import BacktestSpec, run_backtest
 from .streamer import BacktestStreamer, CandleSeries, MarketSeries
@@ -11,6 +11,7 @@
     "BacktestStreamer",
     "CandleSeries",
     "MarketSeries",
+    "TransactionCosts",
     "iter_dukascopy_candles",
     "read_dukascopy_candles",
     "run_backtest",
diff --git a/tradedesk/execution/backtest/client.py b/tradedesk/execution/backtest/client.py
index 84d60d2..2da5c64 100644
--- a/tradedesk/execution/backtest/client.py
+++ b/tradedesk/execution/backtest/client.py
@@ -1,5 +1,6 @@
 import csv
 import itertools
+import logging
 from dataclasses import dataclass
 from datetime import date
 from pathlib import Path
@@ -20,6 +21,28 @@
     MarketSeries,
 )
 
+log = logging.getLogger(__name__)
+
+
+@dataclass(frozen=True)
+class TransactionCosts:
+    """Optional transaction cost overlays applied on top of bid/ask spread.
+
+    All fields default to zero so existing callers are unaffected.
+
+    Attributes:
+        slippage_points: Fixed adverse slippage per fill in price units.
+        slippage_bps: Proportional slippage in basis points (1 bps = 0.01%).
+        commission_per_fill: Fixed commission charged on every fill (£/€/$).
+        commission_per_round_trip: Fixed commission charged once per closed
+            round trip (£/€/$), at exit.
+    """
+
+    slippage_points: float = 0.0
+    slippage_bps: float = 0.0
+    commission_per_fill: float = 0.0
+    commission_per_round_trip: float = 0.0
+
 
 @dataclass
 class Trade:
@@ -28,6 +51,10 @@ class Trade:
     size: float
     price: float
     timestamp: str | None = None
+    raw_price: float = 0.0
+    spread_cost: float = 0.0
+    slippage_cost: float = 0.0
+    commission_cost: float = 0.0
 
 
 @dataclass
@@ -36,6 +63,10 @@ class Position:
     direction: Direction
     size: float
     entry_price: float
+    raw_entry_price: float = 0.0
+    entry_spread_cost: float = 0.0
+    entry_slippage_cost: float = 0.0
+    entry_commission_cost: float = 0.0
 
 
 class BacktestClient(Client):
@@ -57,6 +88,7 @@ def __init__(
     ):
         self._candle_series = candle_series
         self._market_series = market_series or []
+        self._ask_series: list[CandleSeries] = []
 
         self._history: dict[tuple[str, str], list[Candle]] = {
             (s.instrument, s.period): list(s.candles) for s in candle_series
@@ -66,10 +98,12 @@ def __init__(
         self._closed = False
 
         self._mark_price: dict[str, float] = {}
+        self._ask_price: dict[str, float] = {}
         self.trades: list[Trade] = []
         self.positions: dict[str, Position] = {}
         self.realised_pnl: float = 0.0
         self._current_timestamp: str | None = None
+        self._transaction_costs: TransactionCosts = TransactionCosts()
 
     @classmethod
     def from_history(cls, history: dict[tuple[str, str], list[Candle]]) -> "BacktestClient":
@@ -99,14 +133,17 @@ def from_lazy_sources(
         inst = cls.__new__(cls)
         inst._candle_series = list(candle_series)
         inst._market_series = list(market_series or [])
+        inst._ask_series = []
         inst._history = dict(history)
         inst._started = False
         inst._closed = False
         inst._mark_price = {}
+        inst._ask_price = {}
         inst.trades = []
         inst.positions = {}
         inst.realised_pnl = 0.0
         inst._current_timestamp = None
+        inst._transaction_costs = TransactionCosts()
         return inst
 
     @classmethod
@@ -241,8 +278,18 @@ async def start(self) -> None:
     async def close(self) -> None:
         self._closed = True
 
+    def set_ask_series(self, ask_series: list[CandleSeries]) -> None:
+        """Register ask-side candle series for bid/ask-aware fill pricing."""
+        self._ask_series = list(ask_series)
+
+    def set_transaction_costs(self, tc: TransactionCosts) -> None:
+        """Configure transaction cost overlays (slippage and commission)."""
+        self._transaction_costs = tc
+
     def get_streamer(self) -> BacktestStreamer:
-        return BacktestStreamer(self, self._candle_series, self._market_series)
+        return BacktestStreamer(
+            self, self._candle_series, self._market_series, ask_series=self._ask_series
+        )
 
     def _set_current_timestamp(self, ts: str) -> None:
         self._current_timestamp = ts
@@ -250,6 +297,9 @@ def _set_current_timestamp(self, ts: str) -> None:
     def _set_mark_price(self, instrument: str, price: float) -> None:
         self._mark_price[instrument] = float(price)
 
+    def _set_ask_price(self, instrument: str, price: float) -> None:
+        self._ask_price[instrument] = float(price)
+
     def _get_mark_price(self, instrument: str) -> float:
         if instrument not in self._mark_price:
             raise RuntimeError(f"No mark price available for {instrument} (no data replayed yet)")
@@ -294,6 +344,53 @@ async def get_historical_candles(
         candles = self._history.get((instrument, period), [])
         return candles[-num_points:]
 
+    def _compute_fill_price(
+        self, instrument: str, direction: Direction
+    ) -> tuple[float, float, float, float, float]:
+        """Compute executable fill price with cost decomposition.
+
+        Returns:
+            (fill_price, raw_price, spread_cost, slippage_cost, commission_cost)
+
+        ``fill_price`` is the executable price inclusive of all overlays.
+        ``raw_price`` is the mid price (or bid if ask data unavailable).
+        The three cost fields are in price units, except ``commission_cost``
+        which is the absolute monetary amount for this fill.
+        """
+        bid_price = self._get_mark_price(instrument)
+        ask_price = self._ask_price.get(instrument)
+
+        # Determine executable side
+        if ask_price is not None:
+            # Bid/ask pricing available
+            raw_price = (bid_price + ask_price) / 2
+            if direction == Direction.LONG:
+                exec_price = ask_price  # buy at offer
+            else:
+                exec_price = bid_price  # sell at bid
+            spread_cost = abs(exec_price - raw_price)
+        else:
+            # No ask data — warn if ask series is configured (implies a gap)
+            if self._ask_series:
+                log.warning(
+                    "Missing ask price for %s at %s; using bid price (spread cost = 0)",
+                    instrument,
+                    self._current_timestamp,
+                )
+            raw_price = bid_price
+            exec_price = bid_price
+            spread_cost = 0.0
+
+        # Apply slippage overlay (adverse to the trader)
+        tc = self._transaction_costs
+        slippage = tc.slippage_points + exec_price * tc.slippage_bps / 10_000
+        if direction == Direction.LONG:
+            exec_price += slippage
+        else:
+            exec_price -= slippage
+
+        return exec_price, raw_price, spread_cost, slippage, tc.commission_per_fill
+
     async def place_market_order(
         self,
         instrument: str,
@@ -311,7 +408,9 @@ async def place_market_order(
             raise ValueError("size must be > 0")
 
         _direction = Direction.from_order_side(direction)
-        price = self._get_mark_price(instrument)
+        price, raw_price, spread_cost, slippage_cost, commission_cost = self._compute_fill_price(
+            instrument, _direction
+        )
 
         self.trades.append(
             Trade(
@@ -320,9 +419,16 @@ async def place_market_order(
                 size=float(size),
                 price=price,
                 timestamp=self._current_timestamp,
+                raw_price=raw_price,
+                spread_cost=spread_cost,
+                slippage_cost=slippage_cost,
+                commission_cost=commission_cost,
             )
         )
 
+        # Deduct per-fill commission from realised PnL
+        self.realised_pnl -= commission_cost
+
         # Very simple netting model:
         # - BUY opens/increases LONG, SELL opens/increases SHORT
         # - If opposite direction order arrives, close the entire position if sizes match.
@@ -334,6 +440,10 @@ async def place_market_order(
                 direction=_direction,
                 size=float(size),
                 entry_price=price,
+                raw_entry_price=raw_price,
+                entry_spread_cost=spread_cost,
+                entry_slippage_cost=slippage_cost,
+                entry_commission_cost=commission_cost,
             )
             # Emit PositionOpenedEvent
             await get_dispatcher().publish(
@@ -347,13 +457,12 @@ async def place_market_order(
             )
         else:
             if pos.direction == _direction:
-                # Increase position: weighted avg entry
+                # Increase position: weighted avg entry (costs tracked from initial open only)
                 new_size = pos.size + float(size)
                 pos.entry_price = (pos.entry_price * pos.size + price * float(size)) / new_size
                 pos.size = new_size
             else:
-                # Opposite direction: close (only supports full close or reduce; compute realised
-                # on reduced amount)
+                # Opposite direction: close (only supports full close or reduce)
                 close_size = min(pos.size, float(size))
 
                 # Compute PnL for the closed portion
@@ -362,11 +471,14 @@ async def place_market_order(
                 else:
                     closed_pnl = (pos.entry_price - price) * close_size
 
+                # Deduct per-round-trip commission at close
+                closed_pnl -= self._transaction_costs.commission_per_round_trip
+
                 self.realised_pnl += closed_pnl
 
                 pos.size -= close_size
                 if pos.size <= 0:
-                    # Position fully closed - emit event
+                    # Position fully closed - emit event with full cost decomposition
                     await get_dispatcher().publish(
                         PositionClosedEvent(
                             instrument=instrument,
@@ -377,6 +489,14 @@ async def place_market_order(
                             pnl=closed_pnl,
                             exit_reason=exit_reason or "market_order",
                             timestamp=parse_timestamp(self._current_timestamp or ""),
+                            raw_entry_price=pos.raw_entry_price,
+                            raw_exit_price=raw_price,
+                            entry_spread_cost=pos.entry_spread_cost,
+                            exit_spread_cost=spread_cost,
+                            entry_slippage_cost=pos.entry_slippage_cost,
+                            exit_slippage_cost=slippage_cost,
+                            entry_commission_cost=pos.entry_commission_cost,
+                            exit_commission_cost=commission_cost,
                         )
                     )
                     self.positions.pop(instrument, None)
@@ -388,6 +508,10 @@ async def place_market_order(
                         direction=_direction,
                         size=residual,
                         entry_price=price,
+                        raw_entry_price=raw_price,
+                        entry_spread_cost=spread_cost,
+                        entry_slippage_cost=slippage_cost,
+                        entry_commission_cost=commission_cost,
                     )
                     # Emit PositionOpenedEvent for the new residual position
                     await get_dispatcher().publish(
diff --git a/tradedesk/execution/backtest/runner.py b/tradedesk/execution/backtest/runner.py
index 39e65eb..ed7e1f4 100644
--- a/tradedesk/execution/backtest/runner.py
+++ b/tradedesk/execution/backtest/runner.py
@@ -3,7 +3,8 @@
 
 from __future__ import annotations
 
-from dataclasses import dataclass
+import logging
+from dataclasses import dataclass, field
 from datetime import date
 from pathlib import Path
 from typing import TYPE_CHECKING, Callable
@@ -24,7 +25,11 @@
 )
 from tradedesk.types import Candle
 
-from .client import BacktestClient
+from .client import BacktestClient, TransactionCosts
+from .dukascopy import read_dukascopy_candles
+from .streamer import CandleSeries
+
+log = logging.getLogger(__name__)
 
 
 @dataclass(frozen=True)
@@ -41,6 +46,7 @@ class BacktestSpec:
     size: float = 1.0
     half_spread_adjustment: float = 0.0
     reporting_scale: float = 1.0
+    transaction_costs: TransactionCosts = field(default_factory=TransactionCosts)
 
 
 async def run_backtest(
@@ -75,6 +81,30 @@ async def run_backtest(
     )
     await raw_client.start()
 
+    # Load ask-side candles for bid/ask-aware fill pricing.
+    # Falls back to bid-only fills with a warning when ask data is unavailable.
+    try:
+        ask_candles = read_dukascopy_candles(
+            Path(spec.cache_dir),
+            spec.symbol,
+            spec.period,
+            spec.date_from,
+            spec.date_to,
+            price_side="ask",
+        )
+        raw_client.set_ask_series(
+            [CandleSeries(instrument=spec.instrument, period=spec.period, candles=ask_candles)]
+        )
+    except Exception as exc:
+        log.warning(
+            "Could not load ask candles for %s (%s); bid/ask fill pricing disabled: %s",
+            spec.symbol,
+            spec.period,
+            exc,
+        )
+
+    raw_client.set_transaction_costs(spec.transaction_costs)
+
     # Apply half-spread adjustment if specified (e.g., BID -> MID normalization)
     adj = float(spec.half_spread_adjustment or 0.0)
     if adj:
diff --git a/tradedesk/execution/backtest/streamer.py b/tradedesk/execution/backtest/streamer.py
index ab12462..cd94f02 100644
--- a/tradedesk/execution/backtest/streamer.py
+++ b/tradedesk/execution/backtest/streamer.py
@@ -13,6 +13,14 @@
 log = logging.getLogger(__name__)
 
 
+@dataclass(frozen=True)
+class _AskPriceUpdate:
+    """Internal sentinel: updates ask price on the client without emitting an event."""
+
+    instrument: str
+    close: float
+
+
 def _parse_ts(ts: str) -> datetime:
     # Normalise common variants to something datetime.fromisoformat understands.
     # Accepts:
@@ -66,6 +74,20 @@ def _candle_gen(
         )
 
 
+def _ask_gen(
+    cseries: CandleSeries, seq: Iterator[int]
+) -> Iterator[tuple[datetime, int, _AskPriceUpdate]]:
+    """Yield (timestamp, seq, _AskPriceUpdate) for each ask candle.
+
+    Ask price updates are injected into the priority queue *before* the
+    corresponding bid candle events (lower seq) so that ask prices are always
+    current when a strategy fires after a bid CandleClosedEvent.
+    """
+    for c in cseries.candles:
+        ts = _parse_ts(c.timestamp)
+        yield (ts, next(seq), _AskPriceUpdate(instrument=cseries.instrument, close=c.close))
+
+
 def _market_gen(
     mseries: MarketSeries, seq: Iterator[int]
 ) -> Iterator[tuple[datetime, int, MarketData]]:
@@ -81,6 +103,11 @@ class BacktestStreamer(Streamer):
 
     Replays MarketData and CandleClosedEvent events in timestamp order across all
     series, calling `strategy._handle_event(...)`.
+
+    Ask-side candle series (``ask_series``) are interleaved in the priority queue
+    *before* the corresponding bid candle events so that ask prices are always
+    current when a strategy fires.  Ask candles never emit ``CandleClosedEvent``
+    to the consumer — they only update the client's internal ask price.
     """
 
     def __init__(
@@ -88,10 +115,12 @@ def __init__(
         client: Any,
         candle_series: Iterable[CandleSeries],
         market_series: Iterable[MarketSeries],
+        ask_series: Iterable[CandleSeries] | None = None,
     ) -> None:
         self._client = client
         self._candle_series = list(candle_series)
         self._market_series = list(market_series)
+        self._ask_series = list(ask_series) if ask_series is not None else []
         self._connected = False
 
     async def connect(self) -> None:
@@ -109,11 +138,21 @@ async def run(self, consumer: Any) -> None:
         # CandleClosedEvent or MarketData to be orderable.
         seq = itertools.count()
 
-        streams: list[Iterator[Any]] = [_candle_gen(s, seq) for s in self._candle_series]
+        # Ask streams are added FIRST so their events get lower seq values and
+        # are processed before bid candle events at the same timestamp.
+        streams: list[Iterator[Any]] = [_ask_gen(s, seq) for s in self._ask_series]
+        streams += [_candle_gen(s, seq) for s in self._candle_series]
         streams += [_market_gen(s, seq) for s in self._market_series]
 
+        event_ts: str = ""
+
         try:
             for _, __, event in heapq.merge(*streams):
+                if isinstance(event, _AskPriceUpdate):
+                    # Update ask price only — never forward to consumer.
+                    self._client._set_ask_price(event.instrument, event.close)
+                    continue
+
                 if isinstance(event, MarketData):
                     event_ts = event.timestamp
                     # Mark-to-market uses mid price by default
diff --git a/tradedesk/recording/events.py b/tradedesk/recording/events.py
index 96df19f..761ca55 100644
--- a/tradedesk/recording/events.py
+++ b/tradedesk/recording/events.py
@@ -29,6 +29,15 @@ class PositionClosedEvent(DomainEvent):
     exit_price: float
     pnl: float
     exit_reason: str
+    # Cost decomposition (all default to 0; populated when bid/ask data is available)
+    raw_entry_price: float = 0.0
+    raw_exit_price: float = 0.0
+    entry_spread_cost: float = 0.0
+    exit_spread_cost: float = 0.0
+    entry_slippage_cost: float = 0.0
+    exit_slippage_cost: float = 0.0
+    entry_commission_cost: float = 0.0
+    exit_commission_cost: float = 0.0
 
 
 @event
diff --git a/tradedesk/recording/ledger.py b/tradedesk/recording/ledger.py
index 4453664..8a0ebb8 100644
--- a/tradedesk/recording/ledger.py
+++ b/tradedesk/recording/ledger.py
@@ -29,6 +29,10 @@ def trade_rows_from_trades(trades: list[TradeRecord]) -> list[dict[str, str]]:
             "size": str(t.size),
             "price": str(t.price),
             "reason": t.reason,
+            "raw_price": str(t.raw_price),
+            "spread_cost": str(t.spread_cost),
+            "slippage_cost": str(t.slippage_cost),
+            "commission_cost": str(t.commission_cost),
         }
         for t in trades
     ]
@@ -101,7 +105,20 @@ def write_trades_csv(self, path: Path) -> None:
         path.parent.mkdir(parents=True, exist_ok=True)
         with path.open("w", newline="") as f:
             w = csv.writer(f)
-            w.writerow(["timestamp", "instrument", "direction", "size", "price", "reason"])
+            w.writerow(
+                [
+                    "timestamp",
+                    "instrument",
+                    "direction",
+                    "size",
+                    "price",
+                    "reason",
+                    "raw_price",
+                    "spread_cost",
+                    "slippage_cost",
+                    "commission_cost",
+                ]
+            )
             for t in self.trades:
                 w.writerow(
                     [
@@ -111,6 +128,10 @@ def write_trades_csv(self, path: Path) -> None:
                         round(t.size, 4),
                         t.price,
                         t.reason,
+                        round(t.raw_price, 6) if t.raw_price else "",
+                        round(t.spread_cost, 6) if t.spread_cost else "",
+                        round(t.slippage_cost, 6) if t.slippage_cost else "",
+                        round(t.commission_cost, 4) if t.commission_cost else "",
                     ]
                 )
 
@@ -123,7 +144,8 @@ def write_round_trips_csv(self, path: Path) -> None:
 
         Output schema:
         instrument,direction,entry_ts,exit_ts,entry_price,exit_price,size,pnl,
-        hold_minutes,exit_reason,mfe_points,mae_points,mfe_pnl,mae_pnl
+        hold_minutes,exit_reason,mfe_points,mae_points,mfe_pnl,mae_pnl,
+        raw_entry_price,raw_exit_price,spread_cost,slippage_cost,commission_cost
         """
         path.parent.mkdir(parents=True, exist_ok=True)
 
@@ -141,6 +163,11 @@ def write_round_trips_csv(self, path: Path) -> None:
 
         trips = round_trips_from_fills(trade_rows)
 
+        # Build cost lookup keyed by (instrument, timestamp, direction)
+        cost_lookup: dict[tuple[str, str, str], TradeRecord] = {}
+        for rec in self.trades:
+            cost_lookup[(rec.instrument, rec.timestamp, rec.direction)] = rec
+
         with path.open("w", newline="") as f:
             w = csv.writer(f)
             w.writerow(
@@ -159,6 +186,11 @@ def write_round_trips_csv(self, path: Path) -> None:
                     "mae_points",
                     "mfe_pnl",
                     "mae_pnl",
+                    "raw_entry_price",
+                    "raw_exit_price",
+                    "spread_cost",
+                    "slippage_cost",
+                    "commission_cost",
                 ]
             )
 
@@ -183,6 +215,34 @@ def write_round_trips_csv(self, path: Path) -> None:
                     mfe_pnl_val = round(exc.mfe_pnl, 2)
                     mae_pnl_val = round(exc.mae_pnl, 2)
 
+                # Resolve entry and exit cost records
+                entry_dir = "BUY" if t.direction.value == "long" else "SELL"
+                exit_dir = "SELL" if t.direction.value == "long" else "BUY"
+                entry_rec = cost_lookup.get((t.instrument, t.entry_ts, entry_dir))
+                exit_rec = cost_lookup.get((t.instrument, t.exit_ts, exit_dir))
+
+                raw_entry = (
+                    round(entry_rec.raw_price, 6) if entry_rec and entry_rec.raw_price else ""
+                )
+                raw_exit = (
+                    round(exit_rec.raw_price, 6) if exit_rec and exit_rec.raw_price else ""
+                )
+                rt_spread = round(
+                    (entry_rec.spread_cost if entry_rec else 0.0)
+                    + (exit_rec.spread_cost if exit_rec else 0.0),
+                    6,
+                ) or ""
+                rt_slip = round(
+                    (entry_rec.slippage_cost if entry_rec else 0.0)
+                    + (exit_rec.slippage_cost if exit_rec else 0.0),
+                    6,
+                ) or ""
+                rt_comm = round(
+                    (entry_rec.commission_cost if entry_rec else 0.0)
+                    + (exit_rec.commission_cost if exit_rec else 0.0),
+                    4,
+                ) or ""
+
                 w.writerow(
                     [
                         t.instrument,
@@ -199,6 +259,11 @@ def write_round_trips_csv(self, path: Path) -> None:
                         mae_pts,
                         mfe_pnl_val,
                         mae_pnl_val,
+                        raw_entry,
+                        raw_exit,
+                        rt_spread,
+                        rt_slip,
+                        rt_comm,
                     ]
                 )
 
diff --git a/tradedesk/recording/subscriber.py b/tradedesk/recording/subscriber.py
index dcdfcc4..7d8bdac 100644
--- a/tradedesk/recording/subscriber.py
+++ b/tradedesk/recording/subscriber.py
@@ -185,6 +185,10 @@ async def handle_position_closed(self, event: PositionClosedEvent) -> None:
                 size=event.size,
                 price=event.entry_price,
                 reason="entry",
+                raw_price=event.raw_entry_price,
+                spread_cost=event.entry_spread_cost,
+                slippage_cost=event.entry_slippage_cost,
+                commission_cost=event.entry_commission_cost,
             )
             self.ledger.record_trade(entry_trade)
 
@@ -197,6 +201,10 @@ async def handle_position_closed(self, event: PositionClosedEvent) -> None:
             size=event.size,
             price=event.exit_price,
             reason=event.exit_reason,
+            raw_price=event.raw_exit_price,
+            spread_cost=event.exit_spread_cost,
+            slippage_cost=event.exit_slippage_cost,
+            commission_cost=event.exit_commission_cost,
         )
         self.ledger.record_trade(exit_trade)
 
diff --git a/tradedesk/recording/types.py b/tradedesk/recording/types.py
index 2e5e0bc..4041084 100644
--- a/tradedesk/recording/types.py
+++ b/tradedesk/recording/types.py
@@ -30,8 +30,13 @@ class TradeRecord:
     instrument: str
     direction: str  # "BUY" or "SELL"
     size: float  # stake (e.g. £/point)
-    price: float  # executed price (IG points)
+    price: float  # executable fill price (ask for BUY, bid for SELL, with overlays applied)
     reason: str = ""
+    # Cost decomposition — all zero when bid/ask data is unavailable or bid/ask mode is disabled
+    raw_price: float = 0.0  # mid price at fill time
+    spread_cost: float = 0.0  # half-spread charged (price units)
+    slippage_cost: float = 0.0  # fixed + proportional slippage (price units)
+    commission_cost: float = 0.0  # commission charged (monetary units, e.g. £)
 
 
 @dataclass(frozen=True)