Scanner & Détection Automatisée

Architecture modulaire du système de détection

Le scanner de détection de manipulation est organisé en 5 modules indépendants, chacun spécialisé dans un type de manipulation spécifique. Chaque module calcule un score de suspicion (0-100), et le score composite détermine le niveau d'alerte.

Module 1 : Short Manipulation Score

Module 5 : Social & News Sentiment

Tableau récapitulatif des 30 indicateurs

Implémentation Python du Module Short

# manipulation_scanner/modules/short_module.py
import numpy as np
from dataclasses import dataclass

@dataclass
class ShortMetrics:
    si_pct_float: float    # Short Interest % of Float
    ctb_annual: float      # Cost to Borrow annualisé
    dtc: float             # Days to Cover
    ftd_pct_float: float   # FTD % of Float
    utilization: float     # Utilization Rate
    short_vol_pct: float   # Short Volume %

def score_short_manipulation(m: ShortMetrics) -> dict:
    """Calcule le score de manipulation short (0-100)."""
    scores = {}

    # SI % of Float (weight: 20%)
    scores['si'] = np.clip((m.si_pct_float - 5) / 25 * 100, 0, 100)

    # CTB (weight: 20%)
    scores['ctb'] = np.clip((m.ctb_annual - 5) / 75 * 100, 0, 100)

    # Days to Cover (weight: 15%)
    scores['dtc'] = np.clip((m.dtc - 2) / 6 * 100, 0, 100)

    # FTD % of Float (weight: 15%)
    scores['ftd'] = np.clip(m.ftd_pct_float / 1.0 * 100, 0, 100)

    # Utilization (weight: 15%)
    scores['util'] = np.clip((m.utilization - 50) / 45 * 100, 0, 100)

    # Short Volume % (weight: 15%)
    scores['short_vol'] = np.clip((m.short_vol_pct - 30) / 40 * 100, 0, 100)

    # Weighted composite
    weights = {'si': 0.20, 'ctb': 0.20, 'dtc': 0.15,
               'ftd': 0.15, 'util': 0.15, 'short_vol': 0.15}
    composite = sum(scores[k] * weights[k] for k in weights)

    return {
        'composite': round(composite, 1),
        'breakdown': {k: round(v, 1) for k, v in scores.items()},
        'alert_level': _alert_level(composite)
    }

def _alert_level(score):
    if score >= 90: return 'EMERGENCY'
    if score >= 75: return 'CRITICAL'
    if score >= 50: return 'ALERT'
    if score >= 25: return 'WATCH'
    return 'NORMAL'

# Exemple avec nos cas pratiques :
rzlv = ShortMetrics(si_pct_float=11.5, ctb_annual=1.42, dtc=3.2,
                    ftd_pct_float=0.05, utilization=45, short_vol_pct=42)
rxt = ShortMetrics(si_pct_float=8.44, ctb_annual=76.76, dtc=6.94,
                   ftd_pct_float=0.12, utilization=85, short_vol_pct=58)
mvst = ShortMetrics(si_pct_float=13.5, ctb_annual=0.37, dtc=7.48,
                    ftd_pct_float=0.08, utilization=40, short_vol_pct=45)

# RXT → ALERT (CTB + DTC + Util élevés)
# MVST → WATCH (DTC élevé mais CTB et Util faibles)
# RZLV → NORMAL (métriques toutes modérées)

Sources de données gratuites pour le scanner

Alertes Discord en temps réel

# manipulation_scanner/alerts/discord_alert.py
import aiohttp
import json
from datetime import datetime

DISCORD_WEBHOOK = "https://discord.com/api/webhooks/YOUR_WEBHOOK"

async def send_manipulation_alert(ticker: str, score: dict, module_scores: dict):
    """Envoie une alerte manipulation sur Discord."""
    level = score['alert_level']
    color_map = {
        'EMERGENCY': 0xFF0000, 'CRITICAL': 0xFF6600,
        'ALERT': 0xFFAA00, 'WATCH': 0x3399FF, 'NORMAL': 0x00AA00
    }

    # Construire l'embed Discord
    embed = {
        "title": f"🚨 {level} — {ticker} Manipulation Score: {score['composite']}/100",
        "color": color_map.get(level, 0xFFFFFF),
        "timestamp": datetime.utcnow().isoformat(),
        "fields": [
            {"name": "📊 Short Module", "value": f"{module_scores['short']}/100", "inline": True},
            {"name": "🔒 Dark Pool Module", "value": f"{module_scores['darkpool']}/100", "inline": True},
            {"name": "📈 Options Module", "value": f"{module_scores['options']}/100", "inline": True},
            {"name": "📉 Volume Module", "value": f"{module_scores['volume']}/100", "inline": True},
            {"name": "💬 Social Module", "value": f"{module_scores['social']}/100", "inline": True},
        ],
        "footer": {"text": "Market Watch Manipulation Scanner"}
    }

    payload = {"embeds": [embed]}
    async with aiohttp.ClientSession() as session:
        await session.post(DISCORD_WEBHOOK, json=payload)

Validation historique du scanner