""" 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), }