Stratégie orientée "reconnaissance de profils clients" à partir des flux.
Timeframe
5m
Direction
Long Only
Stoploss
-8.0%
Trailing Stop
Yes
ROI
0m: 6.0%, 60m: 3.0%, 180m: 1.5%, 360m: 0.0%
Interface Version
3
Startup Candles
300
Indicators
3
freqtrade/freqtrade-strategies
author@: lenik
from __future__ import annotations
from functools import reduce
from typing import Dict, List, Tuple
import numpy as np
import pandas as pd
import talib.abstract as ta
from freqtrade.strategy import IStrategy, merge_informative_pair
class ClientFlowAIMarketStrategy(IStrategy):
"""
Stratégie orientée "reconnaissance de profils clients" à partir des flux.
⚠️ Limite importante : Freqtrade ne fournit pas l'identité des clients
ni les carnets d'ordre détaillés. Cette stratégie simule donc un
"profil de client" à partir des micro-indices de flux (pression d'achat
et de vente) dérivés des chandelles, du volume et de la volatilité.
Objectif :
- Surveiller plusieurs timeframes (1s -> 3d si disponible) pour détecter
la cohérence des flux. La récupération 1s dépend de l'exchange/données.
- Regrouper les comportements en profils synthétiques (suiveur de tendance,
retour à la moyenne, breakout). Les profils incohérents sont filtrés.
- Construire un score analytique (type IA légère) pour estimer les zones
de valeur basse (buy) et haute (sell) en cohérence avec les profils.
"""
INTERFACE_VERSION = 3
timeframe = "5m"
startup_candle_count = 300
minimal_roi = {
"0": 0.06,
"60": 0.03,
"180": 0.015,
"360": 0,
}
stoploss = -0.08
trailing_stop = True
trailing_stop_positive = 0.012
trailing_stop_positive_offset = 0.025
trailing_only_offset_is_reached = True
process_only_new_candles = True
use_exit_signal = True
exit_profit_only = False
ignore_roi_if_entry_signal = False
plot_config = {
"main_plot": {
"value_score": {"color": "green"},
"flow_coherence": {"color": "orange"},
"profile_alignment": {"color": "purple"},
},
"subplots": {
"AI Flow": {
"ai_entry_score": {"color": "blue"},
"ai_exit_score": {"color": "red"},
"incoherence_score": {"color": "gray"},
}
},
}
def informative_pairs(self) -> List[Tuple[str, str]]:
pairs = self.dp.current_whitelist() if self.dp else []
timeframes = ["1s", "1m", "5m", "10m", "15m", "30m", "1h", "1d", "3d"]
return [(pair, tf) for pair in pairs for tf in timeframes if tf != self.timeframe]
@staticmethod
def _zscore(series: pd.Series, window: int) -> pd.Series:
mean = series.rolling(window=window, min_periods=max(2, window // 2)).mean()
std = series.rolling(window=window, min_periods=max(2, window // 2)).std().replace(0, np.nan)
return (series - mean) / std
@staticmethod
def _tanh_clip(series: pd.Series, limit: float = 3.0) -> pd.Series:
return np.tanh(series.clip(-limit, limit))
def _flow_profile(self, dataframe: pd.DataFrame, suffix: str) -> pd.DataFrame:
price_delta = dataframe[f"close{suffix}"] - dataframe[f"open{suffix}"]
spread = (dataframe[f"high{suffix}"] - dataframe[f"low{suffix}"]).replace(0, np.nan)
pressure = price_delta / spread
volume_norm = self._zscore(dataframe[f"volume{suffix}"], 48).fillna(0)
volatility = ta.ATR(
{
"high": dataframe[f"high{suffix}"],
"low": dataframe[f"low{suffix}"],
"close": dataframe[f"close{suffix}"],
},
timeperiod=14,
) / dataframe[f"close{suffix}"].replace(0, np.nan)
volatility = volatility.fillna(0)
dataframe[f"flow_pressure{suffix}"] = self._tanh_clip(pressure.fillna(0))
dataframe[f"flow_volume{suffix}"] = self._tanh_clip(volume_norm)
dataframe[f"flow_volatility{suffix}"] = self._tanh_clip(self._zscore(volatility, 48).fillna(0))
dataframe[f"flow_coherence{suffix}"] = (
0.5 * dataframe[f"flow_pressure{suffix}"]
+ 0.3 * dataframe[f"flow_volume{suffix}"]
- 0.2 * dataframe[f"flow_volatility{suffix}"]
)
momentum = ta.ROC(dataframe[f"close{suffix}"], timeperiod=9).fillna(0)
reversion = (dataframe[f"close{suffix}"] - ta.SMA(dataframe[f"close{suffix}"], timeperiod=20)).fillna(0)
range_pos = (
(dataframe[f"close{suffix}"] - dataframe[f"low{suffix}"])
/ (spread.replace(0, np.nan))
).fillna(0.5)
dataframe[f"profile_trend{suffix}"] = self._tanh_clip(
self._zscore(momentum, 48) + dataframe[f"flow_pressure{suffix}"]
)
dataframe[f"profile_reversion{suffix}"] = self._tanh_clip(
-self._zscore(reversion, 48) + (0.5 - range_pos)
)
dataframe[f"profile_breakout{suffix}"] = self._tanh_clip(
dataframe[f"flow_coherence{suffix}"] + self._zscore(volatility, 48).fillna(0)
)
dataframe[f"incoherence_score{suffix}"] = self._tanh_clip(
self._zscore(abs(dataframe[f"flow_pressure{suffix}"]), 48).fillna(0)
- dataframe[f"flow_volume{suffix}"]
)
return dataframe
def populate_indicators(self, dataframe: pd.DataFrame, metadata: Dict) -> pd.DataFrame:
dataframe = self._flow_profile(dataframe, "")
if self.dp:
for tf in ["1s", "1m", "10m", "15m", "30m", "1h", "1d", "3d"]:
try:
info = self.dp.get_pair_dataframe(metadata["pair"], tf)
except Exception:
continue
info = self._flow_profile(info, "")
dataframe = merge_informative_pair(dataframe, info, self.timeframe, tf, ffill=True)
dataframe["value_score"] = self._tanh_clip(
self._zscore(dataframe["close"], 96)
+ self._zscore(dataframe["flow_coherence"], 96).fillna(0)
- dataframe["incoherence_score"].fillna(0)
)
flow_columns = [
"flow_coherence_1s",
"flow_coherence_1m",
"flow_coherence_10m",
"flow_coherence_15m",
"flow_coherence_30m",
"flow_coherence_1h",
"flow_coherence_1d",
"flow_coherence_3d",
]
flow_available = [col for col in flow_columns if col in dataframe]
if flow_available:
dataframe["flow_coherence"] = reduce(
lambda left, right: left + right,
[dataframe[col].fillna(0) for col in flow_available],
) / len(flow_available)
incoherence_columns = [
"incoherence_score_1s",
"incoherence_score_1m",
"incoherence_score_10m",
"incoherence_score_15m",
"incoherence_score_30m",
"incoherence_score_1h",
"incoherence_score_1d",
"incoherence_score_3d",
]
incoherence_available = [col for col in incoherence_columns if col in dataframe]
if incoherence_available:
dataframe["incoherence_score"] = reduce(
lambda left, right: left + right,
[dataframe[col].fillna(0) for col in incoherence_available],
) / len(incoherence_available)
profile_columns = [
"profile_trend_1s",
"profile_trend_1m",
"profile_trend_10m",
"profile_trend_15m",
"profile_trend_30m",
"profile_trend_1h",
"profile_trend_1d",
"profile_trend_3d",
]
reversion_columns = [
"profile_reversion_1s",
"profile_reversion_1m",
"profile_reversion_10m",
"profile_reversion_15m",
"profile_reversion_30m",
"profile_reversion_1h",
"profile_reversion_1d",
"profile_reversion_3d",
]
breakout_columns = [
"profile_breakout_1s",
"profile_breakout_1m",
"profile_breakout_10m",
"profile_breakout_15m",
"profile_breakout_30m",
"profile_breakout_1h",
"profile_breakout_1d",
"profile_breakout_3d",
]
def _avg_profile(cols: List[str]) -> pd.Series:
available = [col for col in cols if col in dataframe]
if not available:
return pd.Series(index=dataframe.index, data=0.0)
return (
reduce(lambda left, right: left + right, [dataframe[col].fillna(0) for col in available])
/ len(available)
)
dataframe["profile_trend"] = _avg_profile(profile_columns)
dataframe["profile_reversion"] = _avg_profile(reversion_columns)
dataframe["profile_breakout"] = _avg_profile(breakout_columns)
dataframe["profile_alignment"] = self._tanh_clip(
dataframe["profile_trend"] + dataframe["profile_reversion"] + dataframe["profile_breakout"]
)
dataframe["ai_entry_score"] = self._tanh_clip(
-dataframe["value_score"]
+ dataframe["flow_coherence"].fillna(0)
+ dataframe["profile_alignment"].fillna(0)
)
dataframe["ai_exit_score"] = self._tanh_clip(
dataframe["value_score"]
- dataframe["flow_coherence"].fillna(0)
- dataframe["profile_alignment"].fillna(0)
)
return dataframe
def populate_entry_trend(self, dataframe: pd.DataFrame, metadata: Dict) -> pd.DataFrame:
dataframe.loc[
(
(dataframe["ai_entry_score"] > 0.4)
& (dataframe["flow_coherence"] > 0.05)
& (dataframe["profile_alignment"] > 0.05)
& (dataframe["incoherence_score"] < 0.2)
& (dataframe["volume"] > 0)
),
"enter_long",
] = 1
return dataframe
def populate_exit_trend(self, dataframe: pd.DataFrame, metadata: Dict) -> pd.DataFrame:
dataframe.loc[
(
(dataframe["ai_exit_score"] > 0.35)
| (dataframe["flow_coherence"] < -0.1)
| (dataframe["incoherence_score"] > 0.35)
),
"exit_long",
] = 1
return dataframe