Simplified Market Microstructure & Order Flow Strategy
Timeframe
15m
Direction
Long & Short
Stoploss
-5.0%
Trailing Stop
Yes
ROI
0m: 8.0%, 30m: 4.0%, 90m: 2.0%, 180m: 1.0%
Interface Version
3
Startup Candles
N/A
Indicators
9
freqtrade/freqtrade-strategies
Strategy 003 author@: Gerald Lonlas github@: https://github.com/freqtrade/freqtrade-strategies
# --- Do not remove these imports ---
from functools import reduce
from typing import Dict, Optional
import numpy as np
import pandas as pd
import talib.abstract as ta
from pandas import DataFrame
from freqtrade.optimize.space import Categorical, Dimension, Integer, SKDecimal
from freqtrade.strategy import (
CategoricalParameter,
DecimalParameter,
IntParameter,
IStrategy,
)
class MarketMicrostructureFlow(IStrategy):
"""
Simplified Market Microstructure & Order Flow Strategy
Core concepts:
- Volume delta analysis (simplified)
- ATR-based adaptive bands
- Trend detection with EMAs
- RSI and MACD for momentum
Optimized for 15m futures trading
"""
# Strategy interface version
INTERFACE_VERSION = 3
# Optimal timeframe for the strategy
timeframe = "15m"
# Can short - futures compatible
can_short = True
# ROI table - conservative for futures
minimal_roi = {"0": 0.08, "30": 0.04, "90": 0.02, "180": 0.01}
# Stoploss
stoploss = -0.05 # Tighter stop for futures
# Trailing stop configuration
trailing_stop = True
trailing_stop_positive = 0.01
trailing_stop_positive_offset = 0.02
trailing_only_offset_is_reached = True
# Use exit signal
use_exit_signal = True
exit_profit_only = False
ignore_roi_if_entry_signal = False
# Number of candles the strategy requires before producing valid signals
startup_candle_count: int = 100
# Optional order types mapping
order_types = {
"entry": "limit",
"exit": "limit",
"stoploss": "market",
"stoploss_on_exchange": True,
}
# Optional order time in force
order_time_in_force = {"entry": "gtc", "exit": "gtc"}
# Hyperparameters
# Volume Delta Parameters
delta_period = IntParameter(5, 20, default=10, space="buy")
delta_threshold = DecimalParameter(0.1, 0.5, default=0.25, decimals=2, space="buy")
# ATR Bands Parameters
atr_period = IntParameter(10, 30, default=14, space="buy")
band_mult = DecimalParameter(1.5, 3.0, default=2.0, decimals=1, space="buy")
# Trend Parameters
ema_fast = IntParameter(5, 15, default=9, space="buy")
ema_slow = IntParameter(20, 50, default=21, space="buy")
ema_baseline = IntParameter(50, 100, default=50, space="buy")
# Momentum Parameters
rsi_period = IntParameter(10, 20, default=14, space="buy")
rsi_buy = IntParameter(25, 45, default=35, space="buy")
rsi_sell = IntParameter(55, 75, default=65, space="buy")
# MACD Parameters
macd_fast = IntParameter(8, 16, default=12, space="buy")
macd_slow = IntParameter(20, 30, default=26, space="buy")
macd_signal = IntParameter(7, 12, default=9, space="buy")
# Entry Thresholds
min_trend_strength = DecimalParameter(0.0, 1.0, default=0.3, decimals=1, space="buy")
volume_factor = DecimalParameter(0.8, 2.0, default=1.2, decimals=1, space="buy")
# Exit Parameters
take_profit = DecimalParameter(0.01, 0.05, default=0.02, decimals=3, space="sell")
atr_exit_mult = DecimalParameter(1.0, 3.0, default=1.5, decimals=1, space="sell")
def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
"""
Populate all indicators - simplified version
"""
# Volume Delta - Simple approximation
# Buy pressure when close > (high + low) / 2
mid_point = (dataframe["high"] + dataframe["low"]) / 2
# Calculate buy and sell volume estimates
buy_ratio = (dataframe["close"] - dataframe["low"]) / (
dataframe["high"] - dataframe["low"] + 0.0001
)
buy_ratio = buy_ratio.clip(0, 1)
dataframe["buy_volume"] = dataframe["volume"] * buy_ratio
dataframe["sell_volume"] = dataframe["volume"] * (1 - buy_ratio)
dataframe["volume_delta"] = dataframe["buy_volume"] - dataframe["sell_volume"]
# Smooth volume delta
dataframe["delta_ma"] = (
dataframe["volume_delta"].rolling(window=self.delta_period.value).mean()
)
dataframe["delta_std"] = (
dataframe["volume_delta"].rolling(window=self.delta_period.value).std()
)
# Delta normalized (z-score)
dataframe["delta_zscore"] = (dataframe["volume_delta"] - dataframe["delta_ma"]) / (
dataframe["delta_std"] + 0.0001
)
# ATR and Adaptive Bands
dataframe["atr"] = ta.ATR(dataframe, timeperiod=self.atr_period.value)
# Use EMA as middle line
dataframe["middle_band"] = ta.EMA(dataframe, timeperiod=self.atr_period.value)
dataframe["upper_band"] = dataframe["middle_band"] + (
dataframe["atr"] * self.band_mult.value
)
dataframe["lower_band"] = dataframe["middle_band"] - (
dataframe["atr"] * self.band_mult.value
)
# Band position (0 = at lower, 1 = at upper)
band_range = dataframe["upper_band"] - dataframe["lower_band"]
dataframe["band_position"] = (dataframe["close"] - dataframe["lower_band"]) / (
band_range + 0.0001
)
dataframe["band_position"] = dataframe["band_position"].clip(0, 1)
# EMAs for trend
dataframe["ema_fast"] = ta.EMA(dataframe, timeperiod=self.ema_fast.value)
dataframe["ema_slow"] = ta.EMA(dataframe, timeperiod=self.ema_slow.value)
dataframe["ema_baseline"] = ta.EMA(dataframe, timeperiod=self.ema_baseline.value)
# Trend strength (-1 to 1)
dataframe["trend"] = 0
dataframe.loc[dataframe["ema_fast"] > dataframe["ema_slow"], "trend"] = 1
dataframe.loc[dataframe["ema_fast"] < dataframe["ema_slow"], "trend"] = -1
# RSI
dataframe["rsi"] = ta.RSI(dataframe, timeperiod=self.rsi_period.value)
# MACD
macd = ta.MACD(
dataframe,
fastperiod=self.macd_fast.value,
slowperiod=self.macd_slow.value,
signalperiod=self.macd_signal.value,
)
dataframe["macd"] = macd["macd"]
dataframe["macdsignal"] = macd["macdsignal"]
dataframe["macdhist"] = macd["macdhist"]
# Volume metrics
dataframe["volume_ma"] = dataframe["volume"].rolling(window=20).mean()
dataframe["volume_ratio"] = dataframe["volume"] / dataframe["volume_ma"]
# OBV (On Balance Volume)
dataframe["obv"] = ta.OBV(dataframe)
dataframe["obv_ma"] = dataframe["obv"].rolling(window=20).mean()
dataframe["obv_trend"] = dataframe["obv"] > dataframe["obv_ma"]
# CMF (Chaikin Money Flow)
mfm = (
(dataframe["close"] - dataframe["low"]) - (dataframe["high"] - dataframe["close"])
) / (dataframe["high"] - dataframe["low"] + 0.0001)
mfv = mfm * dataframe["volume"]
dataframe["cmf"] = mfv.rolling(window=20).sum() / (
dataframe["volume"].rolling(window=20).sum() + 0.0001
)
# VWAP
typical_price = (dataframe["high"] + dataframe["low"] + dataframe["close"]) / 3
cum_volume = dataframe["volume"].cumsum()
cum_pv = (typical_price * dataframe["volume"]).cumsum()
dataframe["vwap"] = cum_pv / (cum_volume + 0.0001)
# Distance from VWAP
dataframe["vwap_distance"] = (dataframe["close"] - dataframe["vwap"]) / dataframe["close"]
# Bollinger Bands for additional reference
bb = ta.BBANDS(dataframe, timeperiod=20, nbdevup=2.0, nbdevdn=2.0)
dataframe["bb_lower"] = bb["lowerband"]
dataframe["bb_upper"] = bb["upperband"]
dataframe["bb_middle"] = bb["middleband"]
return dataframe
def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
"""
Populate entry signals - simplified logic
"""
# Long Entry Conditions
long_conditions = []
# 1. Price at or near lower band
long_conditions.append(
(dataframe["close"] <= dataframe["lower_band"] * 1.01)
| (dataframe["band_position"] < 0.2)
)
# 2. Positive volume delta or recovering
long_conditions.append(
(dataframe["delta_ma"] > 0) | (dataframe["delta_zscore"] > self.delta_threshold.value)
)
# 3. RSI not oversold but not overbought
long_conditions.append((dataframe["rsi"] > self.rsi_buy.value) & (dataframe["rsi"] < 60))
# 4. Trend is up or neutral
long_conditions.append(dataframe["trend"] >= 0)
# 5. MACD histogram rising
long_conditions.append(dataframe["macdhist"] > dataframe["macdhist"].shift(1))
# 6. Volume above threshold
long_conditions.append(dataframe["volume_ratio"] > self.volume_factor.value)
# Combine conditions (need at least 4 of 6)
if long_conditions:
long_count = reduce(lambda x, y: x.astype(int) + y.astype(int), long_conditions)
dataframe.loc[long_count >= 4, "enter_long"] = 1
# Short Entry Conditions
short_conditions = []
# 1. Price at or near upper band
short_conditions.append(
(dataframe["close"] >= dataframe["upper_band"] * 0.99)
| (dataframe["band_position"] > 0.8)
)
# 2. Negative volume delta
short_conditions.append(
(dataframe["delta_ma"] < 0) | (dataframe["delta_zscore"] < -self.delta_threshold.value)
)
# 3. RSI not overbought but not oversold
short_conditions.append((dataframe["rsi"] < self.rsi_sell.value) & (dataframe["rsi"] > 40))
# 4. Trend is down or neutral
short_conditions.append(dataframe["trend"] <= 0)
# 5. MACD histogram falling
short_conditions.append(dataframe["macdhist"] < dataframe["macdhist"].shift(1))
# 6. Volume above threshold
short_conditions.append(dataframe["volume_ratio"] > self.volume_factor.value)
# Combine conditions (need at least 4 of 6)
if short_conditions:
short_count = reduce(lambda x, y: x.astype(int) + y.astype(int), short_conditions)
dataframe.loc[short_count >= 4, "enter_short"] = 1
return dataframe
def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
"""
Populate exit signals - simplified logic
"""
# Long Exit Conditions
long_exit = (
# Price reached upper band
(dataframe["close"] >= dataframe["upper_band"] * self.atr_exit_mult.value)
|
# RSI overbought
(dataframe["rsi"] > 70)
|
# Strong negative delta
(dataframe["delta_zscore"] < -1.5)
|
# MACD bearish cross
(
(dataframe["macd"] < dataframe["macdsignal"])
& (dataframe["macd"].shift(1) >= dataframe["macdsignal"].shift(1))
)
)
dataframe.loc[long_exit, "exit_long"] = 1
# Short Exit Conditions
short_exit = (
# Price reached lower band
(dataframe["close"] <= dataframe["lower_band"] / self.atr_exit_mult.value)
|
# RSI oversold
(dataframe["rsi"] < 30)
|
# Strong positive delta
(dataframe["delta_zscore"] > 1.5)
|
# MACD bullish cross
(
(dataframe["macd"] > dataframe["macdsignal"])
& (dataframe["macd"].shift(1) <= dataframe["macdsignal"].shift(1))
)
)
dataframe.loc[short_exit, "exit_short"] = 1
return dataframe
def custom_exit(
self,
pair: str,
trade: "Trade",
current_time: "datetime",
current_rate: float,
current_profit: float,
**kwargs,
):
"""
Custom exit logic
"""
dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe)
if len(dataframe) < 1:
return None
last_candle = dataframe.iloc[-1]
# Quick profit take
if current_profit > self.take_profit.value:
if last_candle["volume_ratio"] > 2.0:
return "high_volume_profit"
# Exit on strong delta reversal
if trade.is_short:
if last_candle["delta_zscore"] > 2.0:
return "delta_reversal"
else: # Long trade
if last_candle["delta_zscore"] < -2.0:
return "delta_reversal"
# Exit if too far from VWAP
if abs(last_candle["vwap_distance"]) > 0.03:
if current_profit > 0:
return "vwap_deviation"
return None
def custom_stoploss(
self,
pair: str,
trade: "Trade",
current_time: "datetime",
current_rate: float,
current_profit: float,
**kwargs,
):
"""
Custom stoploss using ATR
"""
dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe)
if len(dataframe) < 1:
return -0.05 # Default stoploss
last_candle = dataframe.iloc[-1]
# ATR-based stop
atr_stop = (last_candle["atr"] * 2.0) / current_rate
# Tighten stop when in profit
if current_profit > 0.02:
return -min(atr_stop, 0.01)
elif current_profit > 0.01:
return -min(atr_stop, 0.02)
else:
return -min(atr_stop, 0.05)
def leverage(
self,
pair: str,
current_time: "datetime",
current_rate: float,
proposed_leverage: float,
max_leverage: float,
entry_tag: Optional[str],
side: str,
**kwargs,
) -> float:
"""
Adjust leverage based on volatility
"""
dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe)
if len(dataframe) < 1:
return 1.0
last_candle = dataframe.iloc[-1]
# Use band width as volatility measure
band_width = (last_candle["upper_band"] - last_candle["lower_band"]) / last_candle[
"middle_band"
]
if band_width < 0.02: # Low volatility
return min(3.0, max_leverage)
elif band_width < 0.04: # Medium volatility
return min(2.0, max_leverage)
else: # High volatility
return min(1.0, max_leverage)