Stream unificado

// Mapa local de cuotas — indexado por ID de cuota, almacena objetos Odds completos del snapshot.
// Los eventos delta se fusionan en este mapa por ID.
const oddsMap = new Map();
let resuming = false; // lo fija `connected` con `resumed: true`
 
// Oportunidades por tipo, cada una indexada por `id`. Un elemento de `:detected` es una
// oportunidad nueva O una versión actualizada de una que ya tienes (mismo `id`).
const oppsByType = { ev: new Map(), arbitrage: new Map(), middles: new Map(), low_hold: new Map() };
 
// Upsert por `id`. Devuelve true solo la primera vez que se ve un `id` —
// alerta con eso, no con cada elemento de `:detected`.
function upsertOpp(type, opp) {
  const isNew = !oppsByType[type].has(opp.id);
  oppsByType[type].set(opp.id, opp); // reemplaza la versión que tenías
  return isNew;
}
 
const eventSource = new EventSource(
  'https://api.sharpapi.io/api/v1/stream?channel=all&league=nba&api_key=YOUR_KEY'
);
 
eventSource.addEventListener('connected', (e) => {
  const { stream_id, channel, resumed } = JSON.parse(e.data);
  // Conserva las cuotas locales solo en una reanudación (`resumed: true`): entonces se
  // reenvían los deltas perdidos en lugar de un snapshot. Cualquier otra conexión va
  // seguida de un snapshot completo — borra ahora. No uses `reconnected`: también es
  // `true` en una reanudación.
  resuming = resumed === true;
  if (!resuming) oddsMap.clear();
  // Las oportunidades nunca se reanudan: cada conexión las reenvía en `snapshot`.
  for (const map of Object.values(oppsByType)) map.clear();
  console.log(`Stream ${stream_id} connected (${channel})`);
});
 
eventSource.addEventListener('snapshot', (e) => {
  const data = JSON.parse(e.data);
  // Un snapshot tras `resumed: true` significa que el servidor abandonó la reanudación:
  // este snapshot reemplaza tu estado.
  if (resuming) {
    oddsMap.clear();
    resuming = false;
  }
  // Los fragmentos de oportunidades llevan `ev` / `arbitrage` / `middles` / `low_hold` en lugar de `odds`
  for (const type of Object.keys(oppsByType)) {
    for (const opp of data[type] ?? []) oppsByType[type].set(opp.id, opp);
  }
  if (!data.odds) return;
  // Almacenar objetos Odds completos indexados por ID
  for (const odd of data.odds) {
    oddsMap.set(odd.id, odd);
  }
  console.log(`Snapshot chunk: ${data.count} odds (${oddsMap.size}/${data.total} total)`);
});
 
eventSource.addEventListener('snapshot:complete', (e) => {
  const { mode } = JSON.parse(e.data); // "resume", "full_resync" o ausente en una primera conexión
  // Un snapshot completo puede llegar sin ningún chunk de `snapshot` — ya nada coincide
  // con tus filtros —, así que el borrado del handler de `snapshot` nunca se ejecutó. La
  // comprobación de `resuming` evita borrar un snapshot que acabas de guardar.
  if (mode === 'full_resync' && resuming) oddsMap.clear();
  resuming = false;
  console.log(`Snapshot complete: ${oddsMap.size} odds loaded`);
});
 
eventSource.addEventListener('odds:update', (e) => {
  const { odds, book } = JSON.parse(e.data);
  // Fusionar deltas compactos en el estado local — solo se envían campos dinámicos
  for (const delta of odds) {
    const existing = oddsMap.get(delta.id);
    if (existing) {
      Object.assign(existing, delta); // Fusionar campos modificados
    } else {
      // Un `id` que aún no tienes llega como fila completa — guárdalo tal cual
      oddsMap.set(delta.id, delta);
    }
  }
  console.log(`${book}: ${odds.length} odds updated`);
});
 
eventSource.addEventListener('odds:removed', (e) => {
  const { ids, book } = JSON.parse(e.data);
  for (const id of ids) {
    oddsMap.delete(id);
  }
  console.log(`${book}: ${ids.length} odds removed`);
});
 
// Cada elemento de `:detected` se guarda con upsert; solo un `id` visto por primera vez se registra como nuevo.
eventSource.addEventListener('ev:detected', (e) => {
  const { opportunities: opps } = JSON.parse(e.data);
  opps.forEach(opp => {
    if (upsertOpp('ev', opp)) console.log(`+EV: ${opp.selection} at ${opp.ev_percentage}%`);
  });
});
 
eventSource.addEventListener('arb:detected', (e) => {
  const { opportunities: arbs } = JSON.parse(e.data);
  arbs.forEach(arb => {
    if (upsertOpp('arbitrage', arb)) console.log(`Arb: ${arb.profit_percent}% profit`);
  });
});
 
eventSource.addEventListener('middles:detected', (e) => {
  const { opportunities: middles } = JSON.parse(e.data);
  middles.forEach(m => {
    if (upsertOpp('middles', m)) console.log(`Middle: ${m.event_name} — EV ${m.expected_value}%`);
  });
});
 
eventSource.addEventListener('low_hold:detected', (e) => {
  const { opportunities: holds } = JSON.parse(e.data);
  holds.forEach(h => {
    if (upsertOpp('low_hold', h)) console.log(`Low hold: ${h.hold_percentage}%`);
  });
});
 
// `:expired` enumera los ids que ya no están. Olvídalos, para que un `id` que
// vuelva más tarde cuente otra vez como nuevo.
for (const prefix of ['ev', 'arb', 'middles', 'low_hold']) {
  eventSource.addEventListener(`${prefix}:expired`, (e) => {
    const { expired, type } = JSON.parse(e.data); // type: 'ev' | 'arbitrage' | 'middles' | 'low_hold'
    for (const id of expired) oppsByType[type].delete(id);
  });
}
 
eventSource.addEventListener('heartbeat', () => {
  console.log('Connection alive');
});
 
eventSource.onerror = () => {
  console.log('Connection lost, auto-reconnecting...');
};

import EventSource from 'eventsource';
 
// Mapa local de cuotas — indexado por ID de cuota
const oddsMap = new Map();
let resuming = false; // lo fija `connected` con `resumed: true`
// Oportunidades indexadas por `id` — `:detected` trae elementos nuevos Y actualizados
const evMap = new Map();
const arbMap = new Map();
 
const es = new EventSource(
  'https://api.sharpapi.io/api/v1/stream?channel=all&sport=basketball&league=nba',
  { headers: { 'X-API-Key': 'YOUR_KEY' } }
);
 
es.addEventListener('connected', (e) => {
  const { stream_id, resumed } = JSON.parse(e.data);
  // Solo una reanudación (`resumed: true`) conserva el estado; si no, llega un snapshot.
  resuming = resumed === true;
  if (!resuming) oddsMap.clear();
  evMap.clear(); // las oportunidades nunca se reanudan — el snapshot las reenvía
  arbMap.clear();
  console.log(`Stream ${stream_id} connected (resumed: ${resumed ?? 'fresh'})`);
});
 
es.addEventListener('snapshot', (e) => {
  // Snapshot tras `resumed: true`: la reanudación se abandonó y el snapshot reemplaza el estado
  if (resuming) {
    oddsMap.clear();
    resuming = false;
  }
  // Los fragmentos de oportunidades llevan `ev` / `arbitrage` / ... en lugar de `odds`
  const { odds, ev, arbitrage } = JSON.parse(e.data);
  for (const odd of odds ?? []) {
    oddsMap.set(odd.id, odd);
  }
  for (const opp of ev ?? []) evMap.set(opp.id, opp);
  for (const arb of arbitrage ?? []) arbMap.set(arb.id, arb);
});
 
es.addEventListener('snapshot:complete', (e) => {
  const { mode } = JSON.parse(e.data); // "resume", "full_resync" o ausente en una primera conexión
  // Un snapshot completo que no coincide con ninguna fila no envía chunks de `snapshot`: borra aquí también
  if (mode === 'full_resync' && resuming) oddsMap.clear();
  resuming = false;
  console.log(`Ready: ${oddsMap.size} odds loaded`);
});
 
es.addEventListener('odds:update', (e) => {
  const { odds, book } = JSON.parse(e.data);
  // Fusionar deltas compactos en los objetos locales completos
  for (const delta of odds) {
    const existing = oddsMap.get(delta.id);
    // Un `id` que aún no tienes llega como fila completa — guárdalo tal cual
    if (existing) Object.assign(existing, delta);
    else oddsMap.set(delta.id, delta);
  }
  console.log(`${book}: ${odds.length} odds updated`);
});
 
es.addEventListener('odds:removed', (e) => {
  const { ids, book } = JSON.parse(e.data);
  for (const id of ids) oddsMap.delete(id);
  console.log(`${book}: ${ids.length} odds removed`);
});
 
es.addEventListener('ev:detected', (e) => {
  const { opportunities } = JSON.parse(e.data);
  for (const opp of opportunities) {
    const isNew = !evMap.has(opp.id);
    evMap.set(opp.id, opp); // upsert — un id conocido es una actualización (nuevo precio / EV%)
    if (isNew) console.log(`+EV: ${opp.selection} — ${opp.ev_percentage}% EV`);
  }
});
 
es.addEventListener('ev:expired', (e) => {
  const { expired } = JSON.parse(e.data);
  for (const id of expired) evMap.delete(id);
  console.log(`EV expired: ${expired.length} opportunities`);
});
 
es.addEventListener('arb:detected', (e) => {
  const { opportunities } = JSON.parse(e.data);
  for (const arb of opportunities) {
    const isNew = !arbMap.has(arb.id);
    arbMap.set(arb.id, arb); // upsert — un id conocido es una actualización (nuevas cuotas de las patas)
    if (isNew) console.log(`Arb: ${arb.profit_percent}% across ${arb.legs.length} legs`);
  }
});
 
es.addEventListener('arb:expired', (e) => {
  const { expired } = JSON.parse(e.data);
  for (const id of expired) arbMap.delete(id);
  console.log(`Arb expired: ${expired.length} opportunities`);
});
 
es.onerror = (err) => {
  console.error('Stream error:', err.message);
};

import sseclient
import requests
import json
 
url = 'https://api.sharpapi.io/api/v1/stream'
params = {
    'channel': 'all',
    'sport': 'basketball',
    'league': 'nba',
}
headers = {'X-API-Key': 'YOUR_KEY'}
 
# Diccionario local de cuotas — indexado por ID de cuota, almacena objetos completos del snapshot
odds_map: dict[str, dict] = {}
resuming = False  # lo fija 'connected' con resumed True
 
# Oportunidades por tipo, cada una indexada por id. Un elemento de `:detected` es una
# oportunidad nueva O una versión actualizada de una que ya tienes (mismo id).
opps_by_type: dict[str, dict[str, dict]] = {
    'ev': {}, 'arbitrage': {}, 'middles': {}, 'low_hold': {},
}
 
def upsert_opp(opp_type: str, opp: dict) -> bool:
    """Upsert por id. True solo la primera vez que se ve un id — alerta con eso."""
    is_new = opp['id'] not in opps_by_type[opp_type]
    opps_by_type[opp_type][opp['id']] = opp  # reemplaza la versión que tenías
    return is_new
 
response = requests.get(url, params=params, headers=headers, stream=True)
client = sseclient.SSEClient(response)
 
for event in client.events():
    data = json.loads(event.data) if event.data else {}
 
    if event.event == 'connected':
        # Solo una reanudación (resumed True) conserva el estado: entonces el servidor
        # reenvía los deltas perdidos. Todo lo demás va seguido de un snapshot completo.
        resuming = data.get('resumed') is True
        if not resuming:
            odds_map.clear()
        # Las oportunidades nunca se reanudan: cada conexión las reenvía en `snapshot`.
        for held in opps_by_type.values():
            held.clear()
        print(f"Stream {data['stream_id']} connected ({data['channel']})")
 
    elif event.event == 'snapshot':
        # Snapshot tras resumed True: el servidor abandonó la reanudación, este
        # snapshot reemplaza tu estado.
        if resuming:
            odds_map.clear()
            resuming = False
        # Los fragmentos de oportunidades llevan 'ev' / 'arbitrage' / 'middles' / 'low_hold' en lugar de 'odds'
        for opp_type, held in opps_by_type.items():
            for opp in data.get(opp_type, []):
                held[opp['id']] = opp
        # Almacenar objetos de cuotas completos indexados por ID
        for odd in data.get('odds', []):
            odds_map[odd['id']] = odd
        print(f"Snapshot chunk: {data['count']} items ({data['offset'] + data['count']}/{data['total']})")
 
    elif event.event == 'snapshot:complete':
        # Un snapshot completo que no coincide con ninguna fila no envía chunks: borra aquí también
        if data.get('mode') == 'full_resync' and resuming:
            odds_map.clear()
        resuming = False
        print(f"Ready: {len(odds_map)} odds loaded")
 
    elif event.event == 'odds:update':
        # Fusionar deltas compactos en el estado local
        for delta in data.get('odds', []):
            existing = odds_map.get(delta['id'])
            if existing:
                existing.update(delta)
            else:
                # Un id que aún no tienes llega como fila completa — guárdalo tal cual
                odds_map[delta['id']] = delta
        print(f"{data['book']}: {data['count']} odds updated")
 
    elif event.event == 'odds:removed':
        for odd_id in data.get('ids', []):
            odds_map.pop(odd_id, None)
        print(f"{data['book']}: {data['count']} odds removed")
 
    # Cada elemento de `:detected` se guarda con upsert; solo un id visto por primera vez se imprime como nuevo.
    elif event.event == 'ev:detected':
        for opp in data['opportunities']:
            if upsert_opp('ev', opp):
                print(f"+EV: {opp['selection']} at {opp['ev_percentage']}%")
 
    elif event.event == 'arb:detected':
        for arb in data['opportunities']:
            if upsert_opp('arbitrage', arb):
                print(f"Arb: {arb['profit_percent']}% profit")
 
    elif event.event == 'middles:detected':
        for m in data['opportunities']:
            if upsert_opp('middles', m):
                print(f"Middle: {m.get('event_name', '?')} — EV {m.get('expected_value', '?')}%")
 
    elif event.event == 'low_hold:detected':
        for h in data['opportunities']:
            if upsert_opp('low_hold', h):
                print(f"Low hold: {h['hold_percentage']}%")
 
    elif event.event in ('ev:expired', 'arb:expired', 'middles:expired', 'low_hold:expired'):
        # 'type' es 'ev', 'arbitrage', 'middles' o 'low_hold'. Olvida los ids,
        # para que un id que vuelva más tarde cuente otra vez como nuevo.
        for opp_id in data['expired']:
            opps_by_type[data['type']].pop(opp_id, None)
        print(f"{data['type']} expired: {len(data['expired'])} opportunities")
 
    elif event.event == 'heartbeat':
        pass  # keepalive silencioso