feat: Phase 4c-bis — CNN image-based (analyse visuelle graphiques chandeliers)
## Nouveaux modules
### src/ml/cnn_image/
- chart_renderer.py : CandlestickImageRenderer — OHLCV → images 128×128 RGB (mplfinance)
Fond #0d1117, bougies vertes/rouges, volume, sans axes, rendu en mémoire
Fallback 2D si mplfinance absent
- cnn_image_model.py : CandlestickCNN — Conv2D 4-blocs (3→32→64→128→256) + AvgPool + Dense(3)
- cnn_image_strategy_model.py : CNNImageStrategyModel — même interface que MLStrategyModel
### src/strategies/cnn_image_driven/
- cnn_image_strategy.py : CNNImageDrivenStrategy(BaseStrategy), SL/TP ATR, seq_len=64
## Modifications
- ensemble_model.py : attach_cnn_image(), poids XGB=0.30/CNN1D=0.30/CNNImage=0.40
- trading.py : POST /train-cnn-image, GET /train-cnn-image/{id}, GET /cnn-image-models
- docker/requirements/api.txt : mplfinance>=0.12.10b0, Pillow>=10.0.0
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Tika
322
src/ml/cnn_image/chart_renderer.py
Normal file
322
src/ml/cnn_image/chart_renderer.py
Normal file
@@ -0,0 +1,322 @@
|
||||
"""
|
||||
CandlestickImageRenderer — Convertit des données OHLCV en images de graphiques.
|
||||
|
||||
Ce module transforme des séquences de bougies OHLCV en images 128×128 RGB
|
||||
qui peuvent être passées à un CNN Conv2D pour l'analyse visuelle des patterns.
|
||||
|
||||
Rendu :
|
||||
- Fond noir (#0d1117), style TradingView
|
||||
- Bougies vertes (#26a69a) pour la hausse, rouges (#ef5350) pour la baisse
|
||||
- Volume en bas de l'image (via mplfinance)
|
||||
- Pas d'axes, pas de labels, pas de titre
|
||||
- Taille fixe : 128×128 pixels, 3 canaux RGB
|
||||
|
||||
Si mplfinance ou PIL ne sont pas disponibles, un rendu de fallback basique
|
||||
encode les données OHLCV numériquement sous forme d'image 2D.
|
||||
"""
|
||||
|
||||
import io
|
||||
import logging
|
||||
from typing import Optional
|
||||
|
||||
import numpy as np
|
||||
import pandas as pd
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# --- Détection optionnelle de mplfinance et PIL ---
|
||||
try:
|
||||
import mplfinance as mpf
|
||||
import matplotlib
|
||||
matplotlib.use('Agg') # Backend non-interactif (pas d'écran requis)
|
||||
import matplotlib.pyplot as plt
|
||||
MPLFINANCE_AVAILABLE = True
|
||||
logger.debug("mplfinance disponible — rendu haute qualité activé")
|
||||
except ImportError:
|
||||
MPLFINANCE_AVAILABLE = False
|
||||
logger.warning("mplfinance non disponible — utilisation du rendu fallback")
|
||||
|
||||
try:
|
||||
from PIL import Image
|
||||
PIL_AVAILABLE = True
|
||||
except ImportError:
|
||||
PIL_AVAILABLE = False
|
||||
logger.warning("Pillow (PIL) non disponible — rendu fallback uniquement")
|
||||
|
||||
|
||||
# Taille cible des images en pixels
|
||||
IMAGE_SIZE = 128
|
||||
|
||||
|
||||
class CandlestickImageRenderer:
|
||||
"""
|
||||
Convertit des données OHLCV en images de graphiques en chandeliers.
|
||||
|
||||
Chaque image est un instantané visuel de `seq_len` bougies consécutives,
|
||||
rendu avec mplfinance (style fond noir, bougies colorées, volume en bas).
|
||||
Le résultat est normalisé en float32 dans [0, 1].
|
||||
|
||||
Args:
|
||||
image_size: Taille carrée de l'image en pixels (défaut 128)
|
||||
"""
|
||||
|
||||
def __init__(self, image_size: int = IMAGE_SIZE):
|
||||
self.image_size = image_size
|
||||
|
||||
# -------------------------------------------------------------------------
|
||||
# Interface publique
|
||||
# -------------------------------------------------------------------------
|
||||
|
||||
def encode(self, df: pd.DataFrame, seq_len: int = 64) -> np.ndarray:
|
||||
"""
|
||||
Encode toutes les fenêtres glissantes du DataFrame en images.
|
||||
|
||||
Produit N = len(df) - seq_len fenêtres, chacune rendue en image
|
||||
128×128 RGB normalisée [0, 1].
|
||||
|
||||
Args:
|
||||
df: DataFrame OHLCV avec colonnes open/high/low/close/volume
|
||||
seq_len: Nombre de bougies par fenêtre (défaut 64)
|
||||
|
||||
Returns:
|
||||
np.ndarray de forme (N, 3, 128, 128), dtype float32, valeurs [0, 1]
|
||||
Retourne un tableau vide (0, 3, 128, 128) si df est trop court.
|
||||
"""
|
||||
df = self._prepare_df(df)
|
||||
n_windows = len(df) - seq_len
|
||||
|
||||
if n_windows <= 0:
|
||||
logger.warning(
|
||||
f"DataFrame trop court ({len(df)} barres) pour seq_len={seq_len}"
|
||||
)
|
||||
return np.zeros((0, 3, self.image_size, self.image_size), dtype=np.float32)
|
||||
|
||||
images = np.zeros(
|
||||
(n_windows, 3, self.image_size, self.image_size), dtype=np.float32
|
||||
)
|
||||
|
||||
for i in range(n_windows):
|
||||
window = df.iloc[i: i + seq_len]
|
||||
try:
|
||||
img = self._render_single(window)
|
||||
images[i] = img
|
||||
except Exception as e:
|
||||
# Fenêtre problématique → on laisse des zéros pour cette position
|
||||
logger.debug(f"Erreur rendu fenêtre {i} : {e}")
|
||||
|
||||
return images
|
||||
|
||||
def encode_last(self, df: pd.DataFrame, seq_len: int = 64) -> np.ndarray:
|
||||
"""
|
||||
Encode uniquement la dernière fenêtre du DataFrame.
|
||||
|
||||
Utilisé pour la prédiction en temps réel : retourne (1, 3, 128, 128).
|
||||
|
||||
Args:
|
||||
df: DataFrame OHLCV
|
||||
seq_len: Nombre de bougies pour la dernière fenêtre (défaut 64)
|
||||
|
||||
Returns:
|
||||
np.ndarray de forme (1, 3, 128, 128), dtype float32, valeurs [0, 1]
|
||||
|
||||
Raises:
|
||||
ValueError si df est trop court pour seq_len bougies
|
||||
"""
|
||||
df = self._prepare_df(df)
|
||||
|
||||
if len(df) < seq_len:
|
||||
raise ValueError(
|
||||
f"DataFrame insuffisant : {len(df)} barres < seq_len={seq_len}"
|
||||
)
|
||||
|
||||
window = df.iloc[-seq_len:]
|
||||
img = self._render_single(window)
|
||||
# Ajouter dimension batch
|
||||
return img[np.newaxis, ...] # (1, 3, H, W)
|
||||
|
||||
# -------------------------------------------------------------------------
|
||||
# Rendu d'une fenêtre
|
||||
# -------------------------------------------------------------------------
|
||||
|
||||
def _render_single(self, df_window: pd.DataFrame) -> np.ndarray:
|
||||
"""
|
||||
Rend une fenêtre OHLCV en image numpy (3, 128, 128).
|
||||
|
||||
Utilise mplfinance si disponible, sinon fallback encodage numérique.
|
||||
|
||||
Args:
|
||||
df_window: DataFrame OHLCV pour une fenêtre (seq_len barres)
|
||||
|
||||
Returns:
|
||||
np.ndarray de forme (3, 128, 128), float32, valeurs [0, 1]
|
||||
"""
|
||||
if MPLFINANCE_AVAILABLE and PIL_AVAILABLE:
|
||||
return self._render_with_mplfinance(df_window)
|
||||
else:
|
||||
return self._render_fallback(df_window)
|
||||
|
||||
def _render_with_mplfinance(self, df_window: pd.DataFrame) -> np.ndarray:
|
||||
"""
|
||||
Rendu haute qualité via mplfinance.
|
||||
|
||||
Style : fond noir (#0d1117), bougies vertes/rouges, volume en bas,
|
||||
aucun axe ni label ni titre. Image 128×128 RGB.
|
||||
"""
|
||||
# Style personnalisé : fond noir, bougies colorées TradingView
|
||||
style = mpf.make_mpf_style(
|
||||
base_mpf_style='nightclouds',
|
||||
marketcolors=mpf.make_marketcolors(
|
||||
up='#26a69a', # vert TradingView (hausse)
|
||||
down='#ef5350', # rouge TradingView (baisse)
|
||||
wick={'up': '#26a69a', 'down': '#ef5350'},
|
||||
edge={'up': '#26a69a', 'down': '#ef5350'},
|
||||
volume={'up': '#26a69a', 'down': '#ef5350'},
|
||||
),
|
||||
facecolor='#0d1117', # Fond noir GitHub-style
|
||||
figcolor='#0d1117',
|
||||
gridcolor='#0d1117',
|
||||
)
|
||||
|
||||
# Rendu en mémoire via BytesIO
|
||||
buf = io.BytesIO()
|
||||
|
||||
fig, axes = mpf.plot(
|
||||
df_window,
|
||||
type='candle',
|
||||
style=style,
|
||||
volume=True,
|
||||
axisoff=True, # Pas d'axes
|
||||
tight_layout=True,
|
||||
returnfig=True,
|
||||
figsize=(1.28, 1.28), # 128 DPI × 1.28 inch = 128 pixels
|
||||
)
|
||||
|
||||
# Suppression de tous les éléments décoratifs
|
||||
for ax in axes:
|
||||
ax.set_axis_off()
|
||||
ax.set_facecolor('#0d1117')
|
||||
for spine in ax.spines.values():
|
||||
spine.set_visible(False)
|
||||
|
||||
fig.savefig(
|
||||
buf,
|
||||
format='png',
|
||||
dpi=100,
|
||||
bbox_inches='tight',
|
||||
pad_inches=0,
|
||||
facecolor='#0d1117',
|
||||
)
|
||||
plt.close(fig)
|
||||
|
||||
buf.seek(0)
|
||||
pil_img = Image.open(buf).convert('RGB')
|
||||
pil_img = pil_img.resize(
|
||||
(self.image_size, self.image_size), Image.LANCZOS
|
||||
)
|
||||
|
||||
# Conversion en array numpy (H, W, 3) → (3, H, W), float32, [0,1]
|
||||
arr = np.array(pil_img, dtype=np.float32) / 255.0
|
||||
arr = arr.transpose(2, 0, 1) # HWC → CHW
|
||||
|
||||
return arr
|
||||
|
||||
def _render_fallback(self, df_window: pd.DataFrame) -> np.ndarray:
|
||||
"""
|
||||
Rendu de secours sans mplfinance : encode les données OHLCV
|
||||
directement en image 2D normalisée.
|
||||
|
||||
Chaque colonne de l'image correspond à une bougie :
|
||||
- Canal 0 (R) : position relative du close dans le range [low, high]
|
||||
- Canal 1 (G) : amplitude de la bougie (high - low normalisé)
|
||||
- Canal 2 (B) : volume normalisé
|
||||
|
||||
L'image est ensuite redimensionnée à image_size × image_size.
|
||||
"""
|
||||
cols = df_window[['open', 'high', 'low', 'close', 'volume']].copy()
|
||||
n = len(cols)
|
||||
|
||||
if n == 0:
|
||||
return np.zeros((3, self.image_size, self.image_size), dtype=np.float32)
|
||||
|
||||
# Normalisation par colonne
|
||||
highs = cols['high'].values.astype(np.float32)
|
||||
lows = cols['low'].values.astype(np.float32)
|
||||
closes = cols['close'].values.astype(np.float32)
|
||||
opens = cols['open'].values.astype(np.float32)
|
||||
vols = cols['volume'].values.astype(np.float32)
|
||||
|
||||
price_range = highs - lows
|
||||
price_range = np.where(price_range == 0, 1e-8, price_range)
|
||||
|
||||
# Canal R : position du close dans le range de la bougie [0, 1]
|
||||
close_pos = (closes - lows) / price_range
|
||||
|
||||
# Canal G : corps de la bougie (|close - open| / range)
|
||||
body = np.abs(closes - opens) / price_range
|
||||
|
||||
# Canal B : volume normalisé [0, 1]
|
||||
vol_max = vols.max()
|
||||
vol_norm = vols / (vol_max if vol_max > 0 else 1.0)
|
||||
|
||||
# Construction image : 3 × 1 × n → 3 × image_size × image_size
|
||||
# On crée une image de height=image_size, width=n puis on redimensionne
|
||||
img = np.zeros((3, 1, n), dtype=np.float32)
|
||||
img[0, 0, :] = close_pos
|
||||
img[1, 0, :] = body
|
||||
img[2, 0, :] = vol_norm
|
||||
|
||||
# Redimensionnement vers (3, image_size, image_size) via répétition
|
||||
# On étire chaque canal sur les deux dimensions
|
||||
img_resized = np.zeros(
|
||||
(3, self.image_size, self.image_size), dtype=np.float32
|
||||
)
|
||||
for c in range(3):
|
||||
# Répétition en hauteur (axe 0 du canal) et interpolation en largeur
|
||||
channel = img[c, 0, :] # (n,)
|
||||
# Interpolation 1D vers image_size
|
||||
x_orig = np.linspace(0, 1, n)
|
||||
x_new = np.linspace(0, 1, self.image_size)
|
||||
channel_resized = np.interp(x_new, x_orig, channel).astype(np.float32)
|
||||
# Étirer sur toute la hauteur
|
||||
img_resized[c] = np.tile(channel_resized, (self.image_size, 1))
|
||||
|
||||
return img_resized
|
||||
|
||||
# -------------------------------------------------------------------------
|
||||
# Utilitaires
|
||||
# -------------------------------------------------------------------------
|
||||
|
||||
@staticmethod
|
||||
def _prepare_df(df: pd.DataFrame) -> pd.DataFrame:
|
||||
"""
|
||||
Normalise le DataFrame : colonnes en minuscules, index DatetimeIndex.
|
||||
|
||||
Args:
|
||||
df: DataFrame OHLCV brut
|
||||
|
||||
Returns:
|
||||
DataFrame nettoyé avec colonnes open/high/low/close/volume
|
||||
"""
|
||||
df = df.copy()
|
||||
df.columns = [c.lower() for c in df.columns]
|
||||
|
||||
# S'assurer que l'index est un DatetimeIndex (requis par mplfinance)
|
||||
if not isinstance(df.index, pd.DatetimeIndex):
|
||||
try:
|
||||
df.index = pd.to_datetime(df.index)
|
||||
except Exception:
|
||||
# Créer un index artificiel si la conversion échoue
|
||||
df.index = pd.date_range(
|
||||
start='2020-01-01', periods=len(df), freq='1h'
|
||||
)
|
||||
|
||||
# Conserver uniquement les colonnes OHLCV
|
||||
required = ['open', 'high', 'low', 'close', 'volume']
|
||||
for col in required:
|
||||
if col not in df.columns:
|
||||
df[col] = 0.0
|
||||
|
||||
df = df[required].dropna(subset=['open', 'high', 'low', 'close'])
|
||||
df = df.ffill().bfill()
|
||||
|
||||
return df
|
||||
Reference in New Issue
Block a user