Unified Stream

// Lokale Quoten-Map — nach Quoten-ID indiziert, speichert vollständige Odds-Objekte aus dem Snapshot.
// Delta-Events werden anhand der ID in diese Map zusammengeführt.
const oddsMap = new Map();
let resuming = false; // von `connected` mit `resumed: true` gesetzt
 
// Opportunities nach Typ, jeweils nach `id` indiziert. Ein `:detected`-Eintrag ist eine neue
// Opportunity ODER eine aktualisierte Version einer bereits gehaltenen (gleiche `id`).
const oppsByType = { ev: new Map(), arbitrage: new Map(), middles: new Map(), low_hold: new Map() };
 
// Upsert nach `id`. Gibt nur beim ersten Auftreten einer `id` true zurück —
// darauf alarmieren, nicht auf jeden `:detected`-Eintrag.
function upsertOpp(type, opp) {
  const isNew = !oppsByType[type].has(opp.id);
  oppsByType[type].set(opp.id, opp); // ersetzt die gehaltene Version
  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);
  // Lokale Quoten nur bei einem Resume behalten (`resumed: true`): dann werden die
  // verpassten Deltas erneut gesendet statt eines Snapshots. Auf jeden anderen
  // Connect folgt ein vollständiger Snapshot — also jetzt leeren. Nicht
  // `reconnected` verwenden: das ist auch bei einem Resume `true`.
  resuming = resumed === true;
  if (!resuming) oddsMap.clear();
  // Opportunities werden nie per Resume fortgesetzt: jeder Connect sendet sie erneut im `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);
  // Ein Snapshot nach `resumed: true` heißt: der Server hat das Resume aufgegeben,
  // dieser Snapshot ersetzt Ihren Status.
  if (resuming) {
    oddsMap.clear();
    resuming = false;
  }
  // Opportunity-Chunks tragen `ev` / `arbitrage` / `middles` / `low_hold` statt `odds`
  for (const type of Object.keys(oppsByType)) {
    for (const opp of data[type] ?? []) oppsByType[type].set(opp.id, opp);
  }
  if (!data.odds) return;
  // Vollständige Odds-Objekte nach ID indiziert speichern
  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" oder bei einem ersten Connect nicht vorhanden
  // Ein vollständiger Snapshot kann ganz ohne `snapshot`-Chunk kommen — nichts passt mehr
  // zu Ihren Filtern —, dann lief das Leeren im `snapshot`-Handler nie. Die `resuming`-
  // Prüfung verhindert, dass hier ein gerade gespeicherter Snapshot gelöscht wird.
  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);
  // Kompakte Deltas in den lokalen Status zusammenführen — nur dynamische Felder werden gesendet
  for (const delta of odds) {
    const existing = oddsMap.get(delta.id);
    if (existing) {
      Object.assign(existing, delta); // Geänderte Felder zusammenführen
    } else {
      // Eine `id`, die Sie noch nicht halten, kommt als vollständige Zeile — so speichern
      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`);
});
 
// Jeder `:detected`-Eintrag wird per Upsert gespeichert; nur eine erstmals gesehene `id` wird als neu geloggt.
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` listet die ids, die weg sind. Vergessen Sie sie, damit eine `id`,
// die später zurückkommt, wieder als neu zählt.
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';
 
// Lokale Quoten-Map — nach Quoten-ID indiziert
const oddsMap = new Map();
let resuming = false; // von `connected` mit `resumed: true` gesetzt
// Opportunities nach `id` indiziert — `:detected` liefert neue UND aktualisierte Einträge
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);
  // Nur ein Resume (`resumed: true`) behält den Status; sonst folgt ein Snapshot.
  resuming = resumed === true;
  if (!resuming) oddsMap.clear();
  evMap.clear(); // Opportunities werden nie fortgesetzt — der Snapshot sendet sie erneut
  arbMap.clear();
  console.log(`Stream ${stream_id} connected (resumed: ${resumed ?? 'fresh'})`);
});
 
es.addEventListener('snapshot', (e) => {
  // Snapshot nach `resumed: true`: das Resume wurde abgebrochen, der Snapshot ersetzt den Status
  if (resuming) {
    oddsMap.clear();
    resuming = false;
  }
  // Opportunity-Chunks tragen `ev` / `arbitrage` / ... statt `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" oder bei einem ersten Connect nicht vorhanden
  // Ein vollständiger Snapshot ohne passende Zeilen sendet keinen `snapshot`-Chunk — hier also auch leeren
  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);
  // Kompakte Deltas in vollständige lokale Objekte zusammenführen
  for (const delta of odds) {
    const existing = oddsMap.get(delta.id);
    // Eine `id`, die Sie noch nicht halten, kommt als vollständige Zeile — so speichern
    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 — eine bekannte id ist ein Update (neuer Preis / 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 — eine bekannte id ist ein Update (neue Leg-Quoten)
    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'}
 
# Lokales odds dict — nach Quoten-ID indiziert, speichert vollständige Odds-Objekte aus dem Snapshot
odds_map: dict[str, dict] = {}
resuming = False  # von 'connected' mit resumed True gesetzt
 
# Opportunities nach Typ, jeweils nach id indiziert. Ein `:detected`-Eintrag ist eine neue
# Opportunity ODER eine aktualisierte Version einer bereits gehaltenen (gleiche id).
opps_by_type: dict[str, dict[str, dict]] = {
    'ev': {}, 'arbitrage': {}, 'middles': {}, 'low_hold': {},
}
 
def upsert_opp(opp_type: str, opp: dict) -> bool:
    """Upsert nach id. True nur beim ersten Auftreten einer id — darauf alarmieren."""
    is_new = opp['id'] not in opps_by_type[opp_type]
    opps_by_type[opp_type][opp['id']] = opp  # ersetzt die gehaltene Version
    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':
        # Nur ein Resume (resumed True) behält den Status: dann sendet der Server die
        # verpassten Deltas erneut. Auf alles andere folgt ein vollständiger Snapshot.
        resuming = data.get('resumed') is True
        if not resuming:
            odds_map.clear()
        # Opportunities werden nie per Resume fortgesetzt: jeder Connect sendet sie erneut im `snapshot`.
        for held in opps_by_type.values():
            held.clear()
        print(f"Stream {data['stream_id']} connected ({data['channel']})")
 
    elif event.event == 'snapshot':
        # Snapshot nach resumed True: der Server hat das Resume aufgegeben, dieser
        # Snapshot ersetzt Ihren Status.
        if resuming:
            odds_map.clear()
            resuming = False
        # Opportunity-Chunks tragen 'ev' / 'arbitrage' / 'middles' / 'low_hold' statt 'odds'
        for opp_type, held in opps_by_type.items():
            for opp in data.get(opp_type, []):
                held[opp['id']] = opp
        # Vollständige Odds-Objekte nach ID indiziert speichern
        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':
        # Ein vollständiger Snapshot ohne passende Zeilen sendet keinen Chunk — hier also auch leeren
        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':
        # Kompakte Deltas in den lokalen Status zusammenführen
        for delta in data.get('odds', []):
            existing = odds_map.get(delta['id'])
            if existing:
                existing.update(delta)
            else:
                # Eine id, die Sie noch nicht halten, kommt als vollständige Zeile — so speichern
                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")
 
    # Jeder `:detected`-Eintrag wird per Upsert gespeichert; nur eine erstmals gesehene id wird als neu ausgegeben.
    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' ist 'ev', 'arbitrage', 'middles' oder 'low_hold'. Die ids vergessen,
        # damit eine später zurückkehrende id wieder als neu zählt.
        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  # silent keepalive