Measuring Money Flow Momentum with Chaikin Oscillator
The Chaikin Oscillator (ADOSC) is a momentum indicator derived from the Accumulation/Distribution Line (ADL). It measures the momentum of money flow by taking the difference between fast and slow EMAs of the ADL.
In trading, ADOSC zero-line crossovers often signal potential bullish or bearish momentum shifts. However, like many oscillators, it can give false signals in sideways markets. To strengthen the approach, we combine it with:
- SMA trend filter to only trade in the dominant trend direction.
- ATR-based stop losses to adjust risk dynamically to volatility.
- Backtrader analyzers for Sharpe ratio, drawdown, returns, SQN, and trade-level performance.
This article provides both the theoretical foundation and a full Python implementation.
Chaikin Oscillator (ADOSC)
The ADOSC is defined as:
\[ ADOSC_t = EMA_{\text{fast}}(ADL_t) - EMA_{\text{slow}}(ADL_t) \]
where \(ADL_t\) is the Accumulation/Distribution Line:
\[ ADL_t = ADL_{t-1} + MFV_t \]
with Money Flow Volume (MFV):
\[ MFV_t = MFM_t \cdot V_t \]
\[ MFM_t = \frac{(C_t - L_t) - (H_t - C_t)}{H_t - L_t} \]
Here
- \(C_t, H_t, L_t\) are close, high, and low prices at time \(t\),
- \(V_t\) is the volume.
The zero-line crossover rule:
- Long entry if \(ADOSC\) crosses above zero.
- Short entry if \(ADOSC\) crosses below zero.
Backtrader Implementation
We define the Backtrader strategy AdoscMomentum:
class AdoscMomentum(bt.Strategy):
"""
Trades based on Chaikin Oscillator (ADOSC) zero line crossovers,
optionally filtered by a long-term moving average trend.
Entry Conditions:
- Long: ADOSC crosses above 0. (Optional: AND Close > SMA)
- Short: ADOSC crosses below 0. (Optional: AND Close < SMA)
Exit Conditions:
- Target: ADOSC crosses back over the zero line against the position.
- Stop Loss: ATR-based stop from entry price.
"""
params = (
('adosc_fast_period', 3),
('adosc_slow_period', 10),
('trend_filter_period', 50), # Period for optional SMA trend filter
('use_trend_filter', True), # Enable/disable the SMA trend filter
('atr_period', 14),
('stop_loss_atr_multiplier', 2.0),
('printlog', True),
)
def log(self, txt, dt=None, doprint=False):
""" Logging function """
if self.params.printlog or doprint:
dt = dt or self.datas[0].datetime.date(0)
print(f'{dt.isoformat()} | {txt}')
def __init__(self):
""" Initialize indicators and variables """
# Data feeds needed for ADOSC and ATR
self.data_close = self.datas[0].close
self.data_open = self.datas[0].open # Not needed by ADOSC/ATR but good practice
self.data_high = self.datas[0].high
self.data_low = self.datas[0].low
self.data_volume = self.datas[0].volume
# --- CORRECTED ADOSC Instantiation ---
# Use TA-Lib wrapper via bt.talib
self.adosc = bt.talib.ADOSC(self.data_high, self.data_low, self.data_close, self.data_volume,
fastperiod=self.params.adosc_fast_period,
slowperiod=self.params.adosc_slow_period)
# --- End Correction ---
# ADOSC Zero Line Crossover detector
self.adosc_cross = btind.CrossOver(self.adosc, 0) # Signal line is ADOSC, plot line is 0
# Optional Trend Filter SMA
if self.params.use_trend_filter:
self.sma_trend = btind.SimpleMovingAverage(self.data_close, # Use close price for SMA
period=self.params.trend_filter_period)
else:
self.sma_trend = None # Explicitly set to None if not used
# ATR for stops
# ATR requires high, low, close - it will use the datas[0] feed by default
self.atr = btind.AverageTrueRange(period=self.params.atr_period)
# Order tracking
self.order = None
self.stop_order = None
self.entry_price = None
# Minimum period check (depends on ADOSC slow period and trend filter)
required_lookback = self.params.adosc_slow_period
if self.params.use_trend_filter:
required_lookback = max(required_lookback, self.params.trend_filter_period)
required_lookback = max(required_lookback, self.params.atr_period)
# ADOSC also needs time for internal EMAs to stabilize (~slow_period + fast_period)
self.min_periods = required_lookback + self.params.adosc_fast_period + 5 # Add generous buffer
def notify_order(self, order):
""" Handle order notifications """
# Reset order if completed, rejected, canceled, etc.
# Also handle stop loss placement upon entry completion
# And handle stop loss cancellation upon target exit completion
if order.status in [order.Submitted, order.Accepted]:
if not order.alive() and self.order == order: self.order = None
return
if order.status == order.Completed:
executed_price = order.executed.price
executed_comm = order.executed.comm
executed_value = order.executed.value
if order.isbuy(): # Could be Long Entry or Short Stop Loss
if self.stop_order and self.stop_order.ref == order.ref: # It was our Short Stop Loss order
self.log(f'STOP LOSS HIT (SHORT), Price: {executed_price:.2f}, Cost: {executed_value:.2f}, Comm: {executed_comm:.2f}')
self.stop_order = None
elif self.position.size > 0 and self.entry_price is None: # It's a Long Entry completion
self.log(f'BUY EXECUTED (LONG ENTRY), Price: {executed_price:.2f}, Cost: {executed_value:.2f}, Comm: {executed_comm:.2f}')
self.entry_price = executed_price
# Place Stop Loss Order
stop_price = self.entry_price - self.atr[0] * self.params.stop_loss_atr_multiplier
self.stop_order = self.sell(exectype=bt.Order.Stop, price=stop_price)
self.log(f'Placed LONG Stop Loss at: {stop_price:.2f}')
else: # Must be target exit completion (closing short)
self.log(f'TARGET EXIT EXECUTED (Closing Short), Price: {executed_price:.2f}, Cost: {executed_value:.2f}, Comm: {executed_comm:.2f}')
if self.stop_order and self.stop_order.alive():
self.cancel(self.stop_order)
self.log(f'Cancelled associated Stop Loss order (ref: {self.stop_order.ref})')
self.stop_order = None
self.entry_price = None
elif order.issell(): # Could be Short Entry or Long Stop Loss or Long Target Exit
if self.stop_order and self.stop_order.ref == order.ref: # It was our Long Stop Loss order
self.log(f'STOP LOSS HIT (LONG), Price: {executed_price:.2f}, Cost: {executed_value:.2f}, Comm: {executed_comm:.2f}')
self.stop_order = None
elif self.position.size < 0 and self.entry_price is None: # It's a Short Entry completion
self.log(f'SELL EXECUTED (SHORT ENTRY), Price: {executed_price:.2f}, Cost: {executed_value:.2f}, Comm: {executed_comm:.2f}')
self.entry_price = executed_price
# Place Stop Loss Order
stop_price = self.entry_price + self.atr[0] * self.params.stop_loss_atr_multiplier
self.stop_order = self.buy(exectype=bt.Order.Stop, price=stop_price)
self.log(f'Placed SHORT Stop Loss at: {stop_price:.2f}')
else: # Must be target exit completion (closing long)
self.log(f'TARGET EXIT EXECUTED (Closing Long), Price: {executed_price:.2f}, Cost: {executed_value:.2f}, Comm: {executed_comm:.2f}')
if self.stop_order and self.stop_order.alive():
self.cancel(self.stop_order)
self.log(f'Cancelled associated Stop Loss order (ref: {self.stop_order.ref})')
self.stop_order = None
self.entry_price = None
# Reset main order tracker if this was the main order completing
if self.order and self.order.ref == order.ref:
self.order = None
elif order.status in [order.Canceled, order.Margin, order.Rejected]:
self.log(f'Order Canceled/Margin/Rejected: Status {order.getstatusname()} Ref: {order.ref}')
if self.order and self.order.ref == order.ref: self.order = None
if self.stop_order and self.stop_order.ref == order.ref: self.stop_order = None
def notify_trade(self, trade):
""" Handle trade notifications """
if not trade.isclosed:
return
self.log(f'OPERATION PROFIT, GROSS {trade.pnl:.2f}, NET {trade.pnlcomm:.2f}')
# Reset entry price after trade is fully closed
self.entry_price = None
def next(self):
""" Core strategy logic """
# self.log(f'Close: {self.data_close[0]:.2f}, ADOSC: {self.adosc[0]:.2f}, Cross: {self.adosc_cross[0]}')
# Check if indicators are ready
if len(self.data_close) < self.min_periods:
return
# Check if an order is pending
if self.order:
return
# --- Exit Logic (ADOSC crossing back over zero) ---
if self.position:
# Exit Long if ADOSC crosses below zero
if self.position.size > 0 and self.adosc_cross[0] < 0:
self.log(f'CLOSE LONG ON TARGET (ADOSC Cross < 0): ADOSC={self.adosc[0]:.2f}')
self.order = self.close() # Place close order
# Exit Short if ADOSC crosses above zero
elif self.position.size < 0 and self.adosc_cross[0] > 0:
self.log(f'CLOSE SHORT ON TARGET (ADOSC Cross > 0): ADOSC={self.adosc[0]:.2f}')
self.order = self.close() # Place close order
return # Don't check entry if we have a position or placed a close order
# --- Entry Logic ---
if not self.position:
# Check Trend Filter (if enabled)
trend_filter_long_ok = (not self.params.use_trend_filter) or \
(self.sma_trend and self.data_close[0] > self.sma_trend[0])
trend_filter_short_ok = (not self.params.use_trend_filter) or \
(self.sma_trend and self.data_close[0] < self.sma_trend[0])
# Long Entry Condition: ADOSC crosses above 0 AND Trend Filter OK
if self.adosc_cross[0] > 0 and trend_filter_long_ok:
self.log(f'LONG ENTRY SIGNAL: ADOSC Cross > 0 (Value={self.adosc[0]:.2f}), Trend Filter OK={trend_filter_long_ok}')
self.order = self.buy() # Place buy order
# Short Entry Condition: ADOSC crosses below 0 AND Trend Filter OK
elif self.adosc_cross[0] < 0 and trend_filter_short_ok:
self.log(f'SHORT ENTRY SIGNAL: ADOSC Cross < 0 (Value={self.adosc[0]:.2f}), Trend Filter OK={trend_filter_short_ok}')
self.order = self.sell() # Place sell order
def stop(self):
""" Strategy finish """
trend_filter_info = f"(SMA {self.params.trend_filter_period})" if self.params.use_trend_filter else "(No Trend Filter)"
self.log(f'(ADOSC {self.params.adosc_fast_period},{self.params.adosc_slow_period}) {trend_filter_info} (ATR {self.params.atr_period},{self.params.stop_loss_atr_multiplier:.1f}) Ending Value {self.broker.getvalue():.2f}', doprint=True)Entry/exit logic is based on ADOSC crossovers and trend filter, with stop orders placed automatically on entry.
Backtesting Setup
Some sample backtests and results:
##
Conclusion
This study shows how the Chaikin Oscillator can be combined with a trend filter and ATR-based stops for a more robust trading strategy. While ADOSC alone may generate frequent false signals, the addition of a trend filter and volatility-based risk management improves performance stability.
Future directions include optimizing fast/slow ADOSC periods, testing across different assets, or combining ADOSC with other momentum indicators to create hybrid systems.