Customer Retention Engine

# Customer Retention Engine — ingestion pipeline
import asyncio
from neo4j import AsyncGraphDatabase
from pydantic import BaseModel

class XGBoost+SHAPRecord(BaseModel):
    entity_id: str
    behavioural_event_logs_score: float
    metadata: dict[str, str]

async def run_pipeline(
    records: list[XGBoost+SHAPRecord],
    uri: str,
    auth: tuple[str, str],
) -> None:
    driver = AsyncGraphDatabase.driver(uri, auth=auth)
    async with driver.session(database="neo4j") as session:
        await session.execute_write(
            _merge_entities,
            [r.model_dump() for r in records],
            technique="Survival Analysis",
        )
    await driver.close()

async def _merge_entities(tx, batch, technique):
    await tx.run("""
        UNWIND $batch AS row
        MERGE (e:Entity {id: row.entity_id})
          ON CREATE SET e.created   = datetime(),
                        e.technique = $technique
          ON MATCH  SET e.updated   = datetime(),
                        e.score     = row.behavioural_event_logs_score
        """, batch=batch, technique=technique)