Almgren–Chriss Execution with Adaptive Order Slices
Code Freqtrade
Summary
This Freqtrade strategy example applies an Almgren–Chriss style execution schedule to position entries and exits. It estimates a parameter called kappa from rolling price variability, candle range, and volume, using it to shape the fraction traded in each slice. The schedule allocates a larger or smaller portion according to the remaining slices and kappa; when kappa approaches zero, the fraction reduces to an even time-weighted schedule. Slices are spaced by a configurable interval, and the code limits the total number of slices.
The example also includes RSI-based long and short entry signals and partial exits, with a stop loss and return-on-investment setting. These signals illustrate strategy plumbing rather than validate the execution model. The code provides no backtest results or comparison against alternative schedules. Its market-impact inputs use candle range and volume as proxies, so actual impact, fill quality, order-book conditions, and parameter suitability remain untested. It should be treated as an implementation example that requires adaptation and evaluation for the target market.
Key ideas
- The example adapts Almgren–Chriss execution by dividing positions into timed order slices.
- A rolling estimate based on price variability, candle range, and volume determines kappa.
- As kappa approaches zero, the next-slice fraction becomes an even time-weighted allocation.
- RSI rules provide sample entries and partial exits, but the code gives no performance evidence.
- Candle-based impact proxies and execution parameters require validation for the intended market.
Tags
Full text
# AlmgrenChrissStrategy.py
```py
from datetime import datetime, timedelta
import math
import pandas as pd
import numpy as np
from freqtrade.exchange import timeframe_to_minutes
from freqtrade.persistence import Trade
from freqtrade.strategy import IStrategy
import logging
logger = logging.getLogger(__name__)
import talib.abstract as ta
class AlmgrenChrissStrategy(IStrategy):
"""
Almgren-Chriss optimal execution strategy.
Balances market impact and volatility risk using adaptive order slices.
The entry and exit signals are examples and should be adapted to your strategy.
Adopted from:
https://github.com/joshuapjacob/almgren-chriss-optimal-execution
- twap_num_slices: desired number of execution slices.
- twap_interval_minutes: time between execution slices.
- vol_window: lookback period used for calculation.
- factor_lambda: risk-aversion parameter.
- eta_volume_fraction: volume fraction used to calibrate temporary market impact.
- gamma_volume_fraction: volume fraction used to calibrate permanent market impact.
"""
timeframe = "15m"
stoploss = -0.10
minimal_roi = {"0": 0.02}
process_only_new_candles = True
startup_candle_count = 30
can_short = True
position_adjustment_enable = True
twap_num_slices = 10
twap_interval_minutes = 1
vol_window = 96
factor_lambda = 0.01
eta_volume_fraction = 0.01
gamma_volume_fraction = 0.1
kappa_default = 0.6
kappa_max = 5.0
def populate_indicators(self, dataframe: pd.DataFrame, metadata: dict) -> pd.DataFrame:
dataframe["rsi"] = ta.RSI(dataframe)
dataframe["kappa"] = self._ac_kappa_series(dataframe, metadata["pair"])
return dataframe
def populate_entry_trend(self, dataframe: pd.DataFrame, metadata: dict) -> pd.DataFrame:
dataframe.loc[
(dataframe["rsi"] < 45) & (dataframe["volume"] > 0),
"enter_long",
] = 1
# Short entry
dataframe.loc[
(dataframe["rsi"] > 55) & (dataframe["volume"] > 0),
"enter_short",
] = 1
return dataframe
def populate_exit_trend(self, dataframe: pd.DataFrame, metadata: dict) -> pd.DataFrame:
return dataframe
def should_exit_partially(self, trade: Trade, current_time: datetime) -> bool:
"""
Determine whether the trade should be partially exited with slices.
This method is only intended to be called from strategy callbacks.
"""
dataframe, _ = self.dp.get_analyzed_dataframe(
trade.pair, self.timeframe
)
if dataframe.empty:
return False
last_candle = dataframe.iloc[-1]
rsi = last_candle["rsi"]
if trade.is_short:
return rsi < 45
return rsi > 55
def _ac_kappa_series(self, dataframe: pd.DataFrame, pair: str) -> pd.Series:
if dataframe.empty:
return pd.Series(dtype=float, index=dataframe.index)
sigma = dataframe["close"].rolling(self.vol_window).std()
avg_spread = (dataframe["high"] - dataframe["low"]).rolling(self.vol_window).mean()
avg_volume = dataframe["volume"].rolling(self.vol_window).mean()
candle_minutes = timeframe_to_minutes(self.timeframe)
tau = self.twap_interval_minutes / candle_minutes
eta = avg_spread / (self.eta_volume_fraction * avg_volume)
gamma = avg_spread / (self.gamma_volume_fraction * avg_volume)
eta_tilde = eta - 0.5 * gamma * tau
eta_tilde = eta_tilde.where(eta_tilde > 0, eta)
sigma_tau = sigma * math.sqrt(tau)
kappa_tilde_sq = (self.factor_lambda * sigma_tau ** 2) / eta_tilde
acosh_arg = (0.5 * kappa_tilde_sq * tau ** 2 + 1.0).clip(lower=1.0)
raw_kappa = np.arccosh(acosh_arg) / tau
return raw_kappa.clip(upper=self.kappa_max).fillna(self.kappa_default)
def _get_kappa(self, pair: str, current_time: datetime | None = None) -> float:
"""
Get kappa value for slices execution.
This method is only intended to be called from strategy callbacks.
"""
if self.dp is None:
return self.kappa_default
dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe)
if dataframe.empty or "kappa" not in dataframe:
return self.kappa_default
value = dataframe["kappa"].iloc[-1]
return self.kappa_default if pd.isna(value) else float(value)
def _ac_next_slice_fraction(self, remaining_slices: int, kappa: float) -> float:
"""
Fraction of the *currently remaining* amount to trade in the next
slice, given `remaining_slices` .
kappa == 0 reduces exactly to 1/remaining_slices plain TWAP.
"""
m = remaining_slices
if m <= 1:
return 1.0
if kappa <= 1e-12:
return 1.0 / m
denominator = math.sinh(kappa * m)
if denominator == 0:
return 1.0 / m
numerator = math.sinh(kappa * (m - 1))
frac = 1.0 - (numerator / denominator)
return min(max(frac, 0.0), 1.0)
def custom_stake_amount(self, pair: str, current_time: datetime, current_rate: float,
proposed_stake: float, min_stake: float | None, max_stake: float,
leverage: float, entry_tag: str | None, side: str,
**kwargs) -> float:
first_fraction = self._ac_next_slice_fraction(self.twap_num_slices, self._get_kappa(pair))
return proposed_stake * first_fraction
def adjust_trade_position(self, trade: Trade, current_time: datetime,
current_rate: float, current_profit: float,
min_stake: float | None, max_stake: float,
current_entry_rate: float, current_exit_rate: float,
current_entry_profit: float, current_exit_profit: float,
**kwargs
) -> float | None | tuple[float | None, str | None]:
if trade.has_open_orders:
return None
filled_entries = trade.select_filled_orders(trade.entry_side)
entry_slices_done = len(filled_entries)
filled_exits = trade.select_filled_orders(trade.exit_side)
exit_slices_done = len(filled_exits)
already_exiting = exit_slices_done > 0
if already_exiting or self.should_exit_partially(trade, current_time):
return self._next_exit_slice(trade, current_time, filled_exits, exit_slices_done)
if entry_slices_done < self.twap_num_slices:
return self._next_entry_slice(trade, current_time, filled_entries, entry_slices_done, min_stake)
return None
def _next_entry_slice(self, trade: Trade, current_time: datetime,
filled_entries: list, slices_done: int, min_stake: float | None
) -> float | None | tuple[float | None, str | None]:
last_fill_time = filled_entries[-1].order_filled_utc if filled_entries else trade.open_date_utc
next_slice_due_at = last_fill_time + timedelta(minutes=self.twap_interval_minutes)
if current_time < next_slice_due_at:
return None
kappa = self._get_kappa(trade.pair)
first_fraction = self._ac_next_slice_fraction(self.twap_num_slices, kappa)
ac_total_stake = filled_entries[0].stake_amount_filled / first_fraction
stake_already_filled = sum(o.stake_amount_filled for o in filled_entries)
remaining_stake = ac_total_stake - stake_already_filled
remaining_slices = self.twap_num_slices - slices_done
if remaining_stake <= 0 or remaining_slices <= 0:
return None
fraction = self._ac_next_slice_fraction(remaining_slices, kappa)
next_slice_stake = remaining_stake * fraction
if next_slice_stake < (min_stake or 0):
return None
return next_slice_stake
def _next_exit_slice(self, trade: Trade, current_time: datetime,
filled_exits: list, slices_done: int
) -> float | None | tuple[float | None, str | None]:
if slices_done >= self.twap_num_slices:
return None
last_fill_time = filled_exits[-1].order_filled_utc if filled_exits else current_time
next_slice_due_at = last_fill_time + timedelta(minutes=self.twap_interval_minutes)
if slices_done > 0 and current_time < next_slice_due_at:
return None
remaining_slices = self.twap_num_slices - slices_done
if remaining_slices <= 1:
return -trade.stake_amount
fraction = self._ac_next_slice_fraction(remaining_slices, self._get_kappa(trade.pair))
slice_stake = trade.stake_amount * fraction
return -slice_stake
```Shown in full with attribution under the source's licence. Licence: GPL-3.0
This summary was written by Stratmill's research agent from the original; it is not a copy of the source.