Opérations : OMS, Exécution & Monitoring de Production

package broker

import (
    "context"
    "time"
)

// BrokerAdapter est l'interface commune pour tous les brokers
// L'agent IA et le moteur de réconciliation n'utilisent QUE cette interface
// Ils sont complètement ignorants d'IB, Binance, ou tout autre broker
type BrokerAdapter interface {
    // Lecture de l'état live
    GetPositions(ctx context.Context) ([]Position, error)
    GetAccountValue(ctx context.Context) (AccountValue, error)
    GetOpenOrders(ctx context.Context) ([]Order, error)

    // Exécution d'ordres
    PlaceOrder(ctx context.Context, req OrderRequest) (Order, error)
    CancelOrder(ctx context.Context, orderID string) error
    CancelAllOrders(ctx context.Context) error

    // Streaming de données de marché (pour les prix live)
    SubscribeTicker(ctx context.Context, symbols []string) (<-chan TickData, error)

    // Info broker
    BrokerName() string
    IsConnected(ctx context.Context) bool
}

// Position représente une position live (indépendante du broker)
type Position struct {
    Symbol      string
    ISIN        string
    Quantity    float64   // négatif = short
    AvgPrice    float64
    MarketValue float64   // en devise du compte
    PnL         float64
    PnLPct      float64
    AsOf        time.Time
}

// OrderRequest est l'ordre généré par le moteur de réconciliation
type OrderRequest struct {
    Symbol    string
    Quantity  float64     // positif = achat, négatif = vente
    OrderType string      // "LIMIT", "MARKET", "MOC" (Market on Close)
    LimitPx   *float64    // nul si OrderType = "MARKET"
    TimeInForce string    // "DAY", "GTC", "IOC"
    Strategy  string     // tag pour l'audit log (ex: "momentum_breakout")
}

// AccountValue agrège la valeur du compte
type AccountValue struct {
    NetLiquidation float64
    BuyingPower    float64
    MarginUsed     float64
    Currency       string
}

// TickData est un prix de marché live streamé
type TickData struct {
    Symbol    string
    Bid       float64
    Ask       float64
    Last      float64
    Volume    int64
    Timestamp time.Time
}

package broker

import (
    "context"
    "fmt"
    "sync"
    "time"

    ibapi "github.com/hadrienk/ibapi"
)

// IBKRAdapter implémente BrokerAdapter pour Interactive Brokers TWS
type IBKRAdapter struct {
    mu      sync.RWMutex
    client  *ibapi.EClient
    wrapper *IBWrapper   // Récepteur des callbacks asynchrones IB
    host    string
    port    int
    clientID int32
}

// IBWrapper implémente ibapi.EWrapper — reçoit toutes les réponses async d'IB
type IBWrapper struct {
    mu          sync.Mutex
    positions   map[string]*Position    // symbol → position live
    orders      map[int64]*Order        // orderID → ordre
    accountVal  AccountValue
    tickers     map[string]chan TickData
}

func NewIBKRAdapter(host string, port int, clientID int32) *IBKRAdapter {
    wrapper := &IBWrapper{
        positions: make(map[string]*Position),
        orders:    make(map[int64]*Order),
        tickers:   make(map[string]chan TickData),
    }
    client := ibapi.NewEClient(wrapper)
    return &IBKRAdapter{
        client:   client,
        wrapper:  wrapper,
        host:     host,
        port:     port,
        clientID: clientID,
    }
}

func (a *IBKRAdapter) Connect(ctx context.Context) error {
    if err := a.client.Connect(a.host, a.port, a.clientID); err != nil {
        return fmt.Errorf("IB TWS connection failed: %w", err)
    }
    // Démarrer la goroutine de lecture des messages IB
    go a.client.Run()

    // Attendre la confirmation de connexion (timeout 10s)
    select {
    case <-a.wrapper.Connected():
        return nil
    case <-time.After(10 * time.Second):
        return fmt.Errorf("IB TWS connection timeout after 10s")
    case <-ctx.Done():
        return ctx.Err()
    }
}

func (a *IBKRAdapter) GetPositions(ctx context.Context) ([]Position, error) {
    // Demander toutes les positions au TWS
    a.client.ReqPositions()

    // Attendre la réponse async (positions arrivées via wrapper.UpdatePortfolio)
    select {
    case <-a.wrapper.PositionsReady():
        a.mu.RLock()
        defer a.mu.RUnlock()
        positions := make([]Position, 0, len(a.wrapper.positions))
        for _, p := range a.wrapper.positions {
            positions = append(positions, *p)
        }
        return positions, nil
    case <-time.After(5 * time.Second):
        return nil, fmt.Errorf("timeout waiting for IB positions")
    case <-ctx.Done():
        return nil, ctx.Err()
    }
}

func (a *IBKRAdapter) PlaceOrder(ctx context.Context, req OrderRequest) (Order, error) {
    // Convertir un OrderRequest générique en ordre IB
    ibOrder := ibapi.Order{
        Action:      "BUY",
        TotalQuantity: req.Quantity,
        OrderType:   req.OrderType,
        Tif:         req.TimeInForce,
    }
    if req.Quantity < 0 {
        ibOrder.Action = "SELL"
        ibOrder.TotalQuantity = -req.Quantity
    }
    if req.LimitPx != nil {
        ibOrder.LmtPrice = *req.LimitPx
    }

    // Contrat IB pour l'action
    contract := ibapi.Contract{
        Symbol:   req.Symbol,
        SecType:  "STK",
        Exchange: "SMART",   // IB Smart Routing
        Currency: "USD",
    }

    orderID := a.client.NextOrderID()
    a.client.PlaceOrder(orderID, contract, ibOrder)

    return Order{
        ExternalID: fmt.Sprintf("IB-%d", orderID),
        Symbol:     req.Symbol,
        Quantity:   req.Quantity,
        Status:     "SUBMITTED",
        SubmittedAt: time.Now(),
    }, nil
}

func (a *IBKRAdapter) BrokerName() string { return "InteractiveBrokers-TWS" }

package broker

import (
    "context"
    "fmt"
    "net/http"
    "encoding/json"
    "crypto/hmac"
    "crypto/sha256"
    "encoding/hex"
    "time"
    "strconv"
)

// BinanceAdapter implémente BrokerAdapter pour Binance Spot
type BinanceAdapter struct {
    apiKey    string
    secretKey string
    baseURL   string     // "https://api.binance.com" ou "https://testnet.binance.vision"
    client    *http.Client
}

func NewBinanceAdapter(apiKey, secretKey string, testnet bool) *BinanceAdapter {
    base := "https://api.binance.com"
    if testnet {
        base = "https://testnet.binance.vision"
    }
    return &BinanceAdapter{
        apiKey:    apiKey,
        secretKey: secretKey,
        baseURL:   base,
        client:    &http.Client{Timeout: 5 * time.Second},
    }
}

// signParams ajoute timestamp + signature HMAC-SHA256 requise par Binance
func (a *BinanceAdapter) signParams(params string) string {
    ts := strconv.FormatInt(time.Now().UnixMilli(), 10)
    body := params + "×tamp=" + ts

    mac := hmac.New(sha256.New, []byte(a.secretKey))
    mac.Write([]byte(body))
    sig := hex.EncodeToString(mac.Sum(nil))

    return body + "&signature=" + sig
}

func (a *BinanceAdapter) PlaceOrder(ctx context.Context, req OrderRequest) (Order, error) {
    side := "BUY"
    qty := req.Quantity
    if req.Quantity < 0 {
        side = "SELL"
        qty = -req.Quantity
    }

    // Binance utilise BTCUSDT, ETHUSDT — pas BTC-USD
    // Normalisation du symbole (notre OMS utilise BTC-USD en interne)
    binanceSymbol := normalizeToBinanceSymbol(req.Symbol)

    params := fmt.Sprintf(
        "symbol=%s&side=%s&type=%s&quantity=%g&timeInForce=%s",
        binanceSymbol, side, req.OrderType, qty, req.TimeInForce,
    )
    if req.LimitPx != nil {
        params += fmt.Sprintf("&price=%.8f", *req.LimitPx)
    }
    params = a.signParams(params)

    httpReq, _ := http.NewRequestWithContext(ctx, "POST",
        a.baseURL+"/api/v3/order?"+params, nil)
    httpReq.Header.Set("X-MBX-APIKEY", a.apiKey)

    resp, err := a.client.Do(httpReq)
    if err != nil {
        return Order{}, fmt.Errorf("binance order failed: %w", err)
    }
    defer resp.Body.Close()

    var result struct {
        OrderID int64  `json:"orderId"`
        Status  string `json:"status"`
        Symbol  string `json:"symbol"`
    }
    if err := json.NewDecoder(resp.Body).Decode(&result); err != nil {
        return Order{}, fmt.Errorf("binance response decode: %w", err)
    }

    return Order{
        ExternalID:  fmt.Sprintf("BNB-%d", result.OrderID),
        Symbol:      req.Symbol,
        Quantity:    req.Quantity,
        Status:      result.Status,
        SubmittedAt: time.Now(),
    }, nil
}

// normalizeToBinanceSymbol : "BTC-USD" → "BTCUSDT", "ETH-USD" → "ETHUSDT"
func normalizeToBinanceSymbol(symbol string) string {
    mapping := map[string]string{
        "BTC-USD": "BTCUSDT", "ETH-USD": "ETHUSDT",
        "SOL-USD": "SOLUSDT", "BNB-USD": "BNBUSDT",
        "XRP-USD": "XRPUSDT",
    }
    if s, ok := mapping[symbol]; ok { return s }
    // Fallback : remplacer "-USD" par "USDT"
    return symbol[:len(symbol)-4] + "USDT"
}

func (a *BinanceAdapter) BrokerName() string { return "Binance-Spot" }

package oms

import (
    "context"
    "fmt"
    "math"

    "securities-oms/internal/broker"
    "securities-oms/internal/db"
    "securities-oms/internal/risk"
)

// ReconciliationEngine réconcilie le portefeuille cible avec le live
type ReconciliationEngine struct {
    adapter  broker.BrokerAdapter
    store    *db.OrderStore
    risk     *risk.RiskController
    config   ReconcileConfig
}

type ReconcileConfig struct {
    MinOrderSizePct float64   // Ignorer si delta < X% de la position actuelle
    MinOrderSizeUSD float64   // Ignorer si delta < X$ de valeur notionnelle
    MaxOrderSizeUSD float64   // Limiter chaque ordre à X$ (éviter les gros impacts marché)
    LimitOrderSlip  float64   // Pct de slippage autorisé pour prix limit (ex: 0.002 = 0.2%)
}

type TargetAllocation struct {
    Symbol         string
    TargetPctNav   float64    // % du NAV alloué à ce titre
    StrategyTag    string
}

// Reconcile exécute un cycle complet de réconciliation
func (e *ReconciliationEngine) Reconcile(ctx context.Context, targets []TargetAllocation) error {
    // 1. Lire le NAV et les positions live
    acctVal, err := e.adapter.GetAccountValue(ctx)
    if err != nil {
        return fmt.Errorf("get account value: %w", err)
    }
    livePositions, err := e.adapter.GetPositions(ctx)
    if err != nil {
        return fmt.Errorf("get positions: %w", err)
    }

    // Index des positions live par symbole
    liveBySymbol := make(map[string]broker.Position)
    for _, p := range livePositions {
        liveBySymbol[p.Symbol] = p
    }

    // 2. Calculer les deltas
    for _, target := range targets {
        targetUSD := target.TargetPctNav * acctVal.NetLiquidation / 100.0

        livePos := liveBySymbol[target.Symbol]
        deltaUSD := targetUSD - livePos.MarketValue

        // 3. Filtrer les micro-deltas
        if math.Abs(deltaUSD) < e.config.MinOrderSizeUSD {
            continue
        }
        if livePos.MarketValue > 0 && math.Abs(deltaUSD)/livePos.MarketValue < e.config.MinOrderSizePct {
            continue
        }

        // 4. Calculer la quantité en nombre d'actions
        // Pour cela, on a besoin du prix live — streamé via SubscribeTicker
        // Simplifié ici : on utilise un prix mid approximatif
        midPrice := e.getMidPrice(ctx, target.Symbol)
        if midPrice <= 0 {
            continue
        }

        qty := deltaUSD / midPrice

        // Cap par taille max d'ordre
        if deltaUSD > e.config.MaxOrderSizeUSD {
            qty = e.config.MaxOrderSizeUSD / midPrice
        } else if deltaUSD < -e.config.MaxOrderSizeUSD {
            qty = -e.config.MaxOrderSizeUSD / midPrice
        }

        // 5. Valider les garde-fous de risque AVANT de soumettre
        orderReq := broker.OrderRequest{
            Symbol:      target.Symbol,
            Quantity:    qty,
            OrderType:   "LIMIT",
            TimeInForce: "DAY",
            Strategy:    target.StrategyTag,
        }
        // Prix limit = mid ± slippage autorisé
        limitPx := midPrice * (1 + e.config.LimitOrderSlip)
        if qty < 0 { limitPx = midPrice * (1 - e.config.LimitOrderSlip) }
        orderReq.LimitPx = &limitPx

        if err := e.risk.Validate(ctx, orderReq, acctVal); err != nil {
            // Garde-fou déclenché → log et skip, pas de panic
            e.store.LogRejection(ctx, orderReq, err.Error())
            continue
        }

        // 6. Soumettre l'ordre
        order, err := e.adapter.PlaceOrder(ctx, orderReq)
        if err != nil {
            e.store.LogError(ctx, orderReq, err.Error())
            continue
        }

        // Audit trail complet dans PostgreSQL
        e.store.InsertOrderEvent(ctx, order, "SUBMITTED")
    }
    return nil
}

func (e *ReconciliationEngine) getMidPrice(ctx context.Context, symbol string) float64 {
    // Implémentation : query Gold Mart pour le dernier close_price
    // En production : prix live depuis le stream WebSocket broker
    return e.store.GetLastClosePrice(ctx, symbol)
}

package risk

import (
    "context"
    "fmt"
    "math"

    "securities-oms/internal/broker"
)

type RiskController struct {
    maxPositionPct  float64   // 0.10 = 10%
    maxSectorPct    float64   // 0.30 = 30%
    maxDailyLossPct float64   // 0.02 = 2%
    maxOrderUSD     float64   // 50000
    blacklist       map[string]bool
}

func NewRiskController() *RiskController {
    return &RiskController{
        maxPositionPct:  0.10,
        maxSectorPct:    0.30,
        maxDailyLossPct: 0.02,
        maxOrderUSD:     50000,
        blacklist:       loadBlacklist(),   // Depuis fichier de config ou DB
    }
}

// Validate vérifie tous les garde-fous pour un ordre donné
// Retourne une erreur décrivant quel guard a été déclenché, nil si OK
func (rc *RiskController) Validate(ctx context.Context, req broker.OrderRequest, acct broker.AccountValue) error {
    // Guard 1 : Blacklist
    if rc.blacklist[req.Symbol] {
        return fmt.Errorf("RISK_BLACKLIST: %s is on the trading blacklist", req.Symbol)
    }

    // Guard 2 : Notional limit par ordre
    notionalUSD := math.Abs(req.Quantity) * (*req.LimitPx)
    if notionalUSD > rc.maxOrderUSD {
        return fmt.Errorf("RISK_ORDER_TOO_LARGE: $%.0f > max $%.0f", notionalUSD, rc.maxOrderUSD)
    }

    // Guard 3 : Position limit (après exécution de cet ordre)
    currentExposure := rc.getCurrentExposure(ctx, req.Symbol)
    projectedExposure := currentExposure + notionalUSD
    projectedPct := projectedExposure / acct.NetLiquidation
    if projectedPct > rc.maxPositionPct {
        return fmt.Errorf("RISK_POSITION_LIMIT: %.1f%% > max %.1f%% for %s",
            projectedPct*100, rc.maxPositionPct*100, req.Symbol)
    }

    // Guard 4 : Daily loss circuit breaker
    dailyPnL := rc.getDailyPnL(ctx)
    dailyPnLPct := dailyPnL / acct.NetLiquidation
    if dailyPnLPct < -rc.maxDailyLossPct {
        return fmt.Errorf("RISK_DAILY_LOSS: Daily P&L %.2f%% exceeds limit %.2f%%",
            dailyPnLPct*100, -rc.maxDailyLossPct*100)
    }

    return nil   // Tous les garde-fous OK
}

func (rc *RiskController) getCurrentExposure(ctx context.Context, symbol string) float64 {
    // Requête PostgreSQL : SUM des positions ouvertes pour ce symbole
    // (implémentation dans db.OrderStore)
    return 0   // Simplifié pour l'exemple
}

func (rc *RiskController) getDailyPnL(ctx context.Context) float64 {
    // P&L journalier depuis orders.order_events
    return 0   // Simplifié pour l'exemple
}

func loadBlacklist() map[string]bool {
    // Charger depuis /etc/securities-oms/blacklist.txt ou table PostgreSQL
    return map[string]bool{
        "MMTLP": true,   // Pink sheet
        "AMTD":  true,   // Delisted
    }
}

-- Vue de monitoring : comparer la dernière réconciliation au NAV actuel
-- Exécuté toutes les heures par un job Nomad de monitoring

WITH latest_reconciliation AS (
    SELECT
        symbol,
        target_pct_nav,
        live_pct_nav,
        ABS(target_pct_nav - live_pct_nav) AS drift_pct,
        reconciled_at
    FROM ops.reconciliation_snapshots
    WHERE reconciled_at = (
        SELECT MAX(reconciled_at)
        FROM ops.reconciliation_snapshots
    )
),
summary AS (
    SELECT
        COUNT(*) AS num_positions,
        SUM(drift_pct) AS total_drift_pct,
        MAX(drift_pct) AS max_drift_pct,
        COUNT(*) FILTER (WHERE drift_pct > 1.0) AS num_breaches,
        MAX(reconciled_at) AS last_reconciled_at,
        NOW() - MAX(reconciled_at) AS age_since_last_recon
    FROM latest_reconciliation
)

SELECT
    *,
    CASE
        WHEN num_breaches > 0             THEN 'ALERT: Position drift > 1% NAV'
        WHEN age_since_last_recon > '2 hours' THEN 'ALERT: Stale reconciliation'
        WHEN total_drift_pct > 5          THEN 'WARN: High aggregate drift'
        ELSE 'OK'
    END AS health_status
FROM summary;

job "oms-monitoring" {
  type      = "batch"
  namespace = "securities-master"
  periodic {
    crons            = ["0 * * * *"]   # Toutes les heures
    prohibit_overlap = true
  }

  group "health-check" {
    task "drift-check" {
      driver = "exec"
      config {
        command = "/usr/local/bin/psql"
        args    = [
          "-h", "${POSTGRES_HOST}",
          "-U", "monitoring_readonly",
          "-d", "securities_master",
          "-f", "/etc/monitoring/drift_check.sql",
          "--csv", "-t"
        ]
      }
      template {
        destination = "local/env"
        env         = true
        data        = <