Stock Data Platform

def _connect_kafka():
    for attempt in range(1, MAX_RETRIES + 1):
        try:
            consumer = KafkaConsumer(
                KAFKA_TOPIC,
                bootstrap_servers=[os.environ.get("KAFKA_BOOTSTRAP", "...:9092")],
                auto_offset_reset="earliest",
                group_id="stock-data-consumer",
                value_deserializer=lambda m: json.loads(m.decode("utf-8")),
            )
            return consumer
        except Exception as e:
            delay = min(BACKOFF_BASE * (2 ** (attempt - 1)), BACKOFF_CAP)
            time.sleep(delay)
    raise ConnectionError(f"Failed to connect after {MAX_RETRIES} attempts")

# batch by size OR time, whichever trips first
if batch and (len(batch) >= BATCH_SIZE or now - last_flush >= FLUSH_INTERVAL):
    conn = _flush_batch(conn, batch)
    batch, last_flush = [], now

def _normalize_columns(df, ticker):
    """Map yfinance columns (with or without ticker suffix) to open/high/low/close/volume."""
    if isinstance(df.columns, pd.MultiIndex):
        df.columns = ["_".join(col) for col in df.columns]
    col_map = {}
    for std in ["Open", "High", "Low", "Close", "Volume"]:
        suffixed = f"{std}_{ticker}"
        if suffixed in df.columns:
            col_map[suffixed] = std.lower()
        elif std in df.columns:
            col_map[std] = std.lower()
    return df.rename(columns=col_map)

# re-read a short overlap from the last loaded date -> reruns are idempotent;
# the first run has no last date and takes the full history
last_date = _get_last_loaded_date(ticker)
if last_date:
    start_date = last_date - timedelta(days=OVERLAP_DAYS)
    df = _normalize_columns(yf.download(ticker, start=start_date, end=end_date, progress=False), ticker)
else:
    df = _normalize_columns(yf.download(ticker, period="max", progress=False), ticker)