Abdoul Ahad Binizi
68 lines
1.9 KiB
Python
68 lines
1.9 KiB
Python
"""
|
|
AHAD QUANT — Order Flow Analyzer (stub)
|
|
Analyse du flux d'ordres / déséquilibre bid-ask.
|
|
|
|
Ce module est un placeholder fonctionnel :
|
|
- Importable sans erreur
|
|
- Retourne des valeurs neutres
|
|
- À compléter avec une vraie implémentation si le broker expose un order book.
|
|
|
|
Pour OANDA : l'API v20 expose /instruments/{instrument}/orderBook
|
|
Pour MT5 : la profondeur de marché est accessible via market_book_get()
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
from typing import Optional
|
|
|
|
|
|
class OrderFlowAnalyzer:
|
|
"""Analyse l'imbalance buy/sell dans le carnet d'ordres."""
|
|
|
|
def __init__(self, depth: int = 20):
|
|
self.depth = depth
|
|
|
|
def get_imbalance(
|
|
self,
|
|
bids: list[tuple[float, float]],
|
|
asks: list[tuple[float, float]],
|
|
) -> float:
|
|
"""
|
|
Retourne un score entre -1 (pression vendeuse) et +1 (pression acheteuse).
|
|
0.0 = neutre / données insuffisantes.
|
|
"""
|
|
if not bids or not asks:
|
|
return 0.0
|
|
|
|
bid_vol = sum(v for _, v in bids[: self.depth])
|
|
ask_vol = sum(v for _, v in asks[: self.depth])
|
|
total = bid_vol + ask_vol
|
|
|
|
if total == 0:
|
|
return 0.0
|
|
|
|
return round((bid_vol - ask_vol) / total, 4)
|
|
|
|
def classify(self, imbalance: float) -> str:
|
|
"""Classe l'imbalance en signal directionnel."""
|
|
if imbalance > 0.25:
|
|
return "BUY_PRESSURE"
|
|
if imbalance < -0.25:
|
|
return "SELL_PRESSURE"
|
|
return "NEUTRAL"
|
|
|
|
|
|
def get_order_flow_signal(
|
|
bids: Optional[list] = None,
|
|
asks: Optional[list] = None,
|
|
) -> dict:
|
|
"""
|
|
Point d'entrée rapide utilisable depuis n'importe quel module.
|
|
Retourne {"imbalance": float, "signal": str}.
|
|
"""
|
|
analyzer = OrderFlowAnalyzer()
|
|
imbalance = analyzer.get_imbalance(bids or [], asks or [])
|
|
return {
|
|
"imbalance": imbalance,
|
|
"signal": analyzer.classify(imbalance),
|
|
}
|