linkbay-analytics 0.2.0
Lightweight e-commerce analytics library with service-based architecture for LinkBay
License: MIT
Requires Python: >=3.8
Keywords: analytics, e-commerce, metrics, fastapi, ai, linkbay

# LinkBay Analytics

**Version**: 0.2.0

Una libreria lightweight per analytics e-commerce con architettura service-based. Nessun modello database incluso: gli utenti forniscono i propri modelli e layer di persistenza. L'obiettivo è essere pragmatici: dependency injection, pure SQL, dipendenze minime.

## Principali capacità

- Architettura service-based con dependency injection: gli utenti iniettano i propri modelli
- Tracking eventi e-commerce (purchase, add_to_cart, product_view, custom events)
- Aggregazioni metriche pure SQL senza Pandas: revenue, conversion rate, AOV, cart abandonment
- Insights AI con fallback rule-based: anomaly detection, trends, recommendations
- Cross-tenant benchmarking con preservazione privacy: percentili p25, p50, p75
- Realtime metrics con Redis opzionale o in-memory fallback
- Router FastAPI configurabile con endpoint condizionali

## Indice

- [Installazione](#installazione)
- [Quick start](#quick-start)
- [Panoramica componenti](#panoramica-componenti)
- [Ricette d'uso](#ricette-duso)
- [Configurazione](#configurazione)
- [Troubleshooting](#troubleshooting)
- [FAQ performance e sicurezza](#faq-performance-e-sicurezza)
- [Licenza](#licenza)

## Installazione

```bash
# Dipendenze core
pip install linkbay-analytics

# Con Redis per caching realtime
pip install "linkbay-analytics[redis]"

# Con AI insights (linkbay-ai)
pip install "linkbay-analytics[ai]"

# Tutte le feature
pip install "linkbay-analytics[redis,ai]"
```

| Requisito | Versione | Note |
|-----------|----------|------|
| Python | 3.8 – 3.12 | testato in CI |
| FastAPI | >= 0.110.0 | routing e dependency injection |
| Pydantic | >= 2.0.0 | validazione schemi API |
| linkbay-core | >= 0.1.0 | utilities ecosistema LinkBay |
| Redis (opz.) | >= 5.0.0 | richiesto solo per RealtimeService con cache esterna |
| linkbay-ai (opz.) | >= 0.2.0 | richiesto solo per insights AI via AIServiceProtocol |

**Servizi esterni**: PostgreSQL 12+ con supporto JSONB (per properties degli eventi).

## Quick start

```python
import asyncio
from sqlalchemy import Column, String, DateTime, JSON
from sqlalchemy.ext.declarative import declarative_base
from linkbay_analytics import (
    EventService,
    MetricsService,
    create_analytics_router,
)

Base = declarative_base()

# 1. Definisci il tuo modello Event
class AnalyticsEvent(Base):
    __tablename__ = "analytics_events"
    id = Column(String, primary_key=True)
    tenant_id = Column(String, nullable=False, index=True)
    event_name = Column(String, nullable=False)
    created_at = Column(DateTime, nullable=False)
    properties = Column(JSON, default={})
    session_id = Column(String)
    user_id = Column(String)

# 2. Crea i servizi iniettando il tuo modello
async def main() -> None:
    from my_app.database import async_session
    
    session = async_session()
    
    event_service = EventService(
        event_model=AnalyticsEvent,
        db_session=session
    )
    
    metrics_service = MetricsService(
        event_model=AnalyticsEvent,
        db_session=session
    )
    
    # 3. Crea il router FastAPI
    router = create_analytics_router(
        event_service=event_service,
        metrics_service=metrics_service,
    )
    
    # 4. Aggiungi al tuo app
    # app.include_router(router, prefix="/api/v1")

asyncio.run(main())
```

**Importante**: La libreria NON include modelli database. Devi creare la tabella `analytics_events` nel tuo schema con i campi sopra elencati. Usa SQLAlchemy, Django ORM, o raw SQL secondo le tue preferenze.

## Panoramica componenti

| Componente | Ruolo | Default |
|------------|-------|---------|
| `EventService` | Tracking eventi con modello utente iniettato | Accetta `event_model` e `db_session` |
| `MetricsService` | Aggregazioni SQL pure (revenue, conversion, AOV) | Usa `event_model` per query native |
| `InsightsService` | Insights AI o rule-based | `ai_service=None` → fallback regole euristiche |
| `BenchmarkService` | Cross-tenant benchmarks con percentili | `min_tenants=5` per privacy |
| `RealtimeService` | Metriche live (active users, today revenue) | `cache=None` → in-memory fallback |
| `create_analytics_router()` | Factory router FastAPI con DI | Endpoint condizionali basati su servizi forniti |

## Ricette d'uso

### 1. Tracking eventi base

```python
from linkbay_analytics.schemas import EventCreate

async def track_purchase(tenant_id: str, order_data: dict):
    event = EventCreate(
        event_name="purchase",
        tenant_id=tenant_id,
        session_id=order_data.get("session_id"),
        user_id=order_data.get("user_id"),
        properties={
            "revenue": 149.99,
            "product_id": "prod_123",
            "quantity": 2,
            "currency": "EUR",
        }
    )
    
    response = await event_service.track_event(event, tenant_id)
    return response  # EventResponse(id=..., created_at=...)
```

**Dettagli rapidi**:
- Il campo `properties` è JSONB: memorizza dati arbitrari come revenue, product_id, ecc.
- `session_id` e `user_id` sono opzionali ma necessari per conversion rate e funnel analysis
- `event_name` può essere qualsiasi stringa: usa `EventType` enum o custom names

### 2. Dashboard metriche aggregate

```python
from linkbay_analytics.schemas import MetricsPeriod

async def get_dashboard(tenant_id: str):
    metrics = await metrics_service.get_dashboard_metrics(
        tenant_id=tenant_id,
        period=MetricsPeriod.LAST_30_DAYS
    )
    
    print(f"Revenue: €{metrics.total_revenue:.2f}")
    print(f"Orders: {metrics.total_orders}")
    print(f"Conversion: {metrics.conversion_rate:.2f}%")
    print(f"AOV: €{metrics.average_order_value:.2f}")
    print(f"Cart Abandonment: {metrics.cart_abandonment_rate:.2f}%")
    
    return metrics
```

**Dettagli rapidi**:
- Le aggregazioni usano pure SQL con `DATE_TRUNC` e `COALESCE` per performance
- Include calcolo automatico di change percentages confrontando con periodo precedente
- Nessun Pandas coinvolto: risparmio ~100MB di dipendenze

### 3. Time series revenue

```python
async def get_revenue_chart(tenant_id: str):
    timeseries = await metrics_service.get_revenue_time_series(
        tenant_id=tenant_id,
        period=MetricsPeriod.LAST_30_DAYS,
        granularity="day"  # o "hour"
    )
    
    for point in timeseries.data_points:
        print(f"{point.date}: €{point.revenue} ({point.orders} orders)")
    
    return timeseries
```

Usa per grafici revenue over time: ogni data point include `revenue`, `orders`, `average_order_value`.

### 4. Top products performance

```python
async def get_best_sellers(tenant_id: str):
    top_products = await metrics_service.get_top_products(
        tenant_id=tenant_id,
        period=MetricsPeriod.LAST_30_DAYS,
        limit=10,
        sort_by="revenue"  # o "quantity", "views"
    )
    
    for product in top_products.products:
        print(f"{product.product_name}: €{product.revenue}")
        print(f"  Sold: {product.quantity_sold}")
        print(f"  Views: {product.views}")
        print(f"  Conversion: {product.conversion_rate:.2f}%")
    
    return top_products
```

**Dettagli rapidi**:
- Estrae `product_id` e `product_name` dal campo JSONB `properties`
- Calcola conversion rate: (purchases / views) * 100
- Query SQL con aggregazioni GROUP BY product_id

### 5. AI insights con fallback

```python
from linkbay_analytics import InsightsService

async def generate_insights(tenant_id: str):
    # Con AI service (opzionale)
    insights_service = InsightsService(
        metrics_service=metrics_service,
        ai_service=None  # O get_ai_service() da linkbay-ai
    )
    
    insights = await insights_service.generate_insights(tenant_id)
    
    for insight in insights.insights:
        print(f"[{insight.severity}] {insight.title}")
        print(f"  {insight.message}")
        print(f"  → {insight.recommendation}")
    
    print(f"Used AI: {insights.used_ai}")
    return insights
```

**Dettagli rapidi**:
- Se `ai_service=None`: usa regole euristiche (cart_abandonment > 70%, conversion < 2%, ecc.)
- Se fornisci `AIServiceProtocol`: chiama `analyze_metrics()` per insights avanzati
- Fallback automatico a rule-based se AI service fallisce

### 6. Cross-tenant benchmarks

```python
from linkbay_analytics import BenchmarkService

async def compare_to_industry(tenant_id: str):
    benchmark_service = BenchmarkService(
        event_model=AnalyticsEvent,
        db_session=session,
        min_tenants=5  # Privacy threshold
    )
    
    benchmarks = await benchmark_service.get_benchmarks(
        tenant_id=tenant_id,
        industry=None,  # Opzionale: filtra per settore
        period=MetricsPeriod.LAST_30_DAYS
    )
    
    for metric in benchmarks.benchmarks:
        print(f"{metric.metric_name}:")
        print(f"  Your value: {metric.your_value}")
        print(f"  Industry p25: {metric.industry_p25}")
        print(f"  Industry median: {metric.industry_median}")
        print(f"  Industry p75: {metric.industry_p75}")
    
    return benchmarks
```

Usa SQL `PERCENTILE_CONT` per calcoli privacy-preserving: minimo 5 tenant per mostrare benchmarks.

### 7. Realtime metrics con Redis

```python
from linkbay_analytics import RealtimeService
from redis.asyncio import Redis

async def setup_realtime():
    # Con Redis
    redis_client = Redis(host='localhost', port=6379)
    realtime_service = RealtimeService(cache=redis_client)
    
    # Track active user
    await realtime_service.track_active_user(
        tenant_id="tenant_123",
        user_id="user_456",
        ttl=300  # 5 minuti
    )
    
    # Get active count
    active = await realtime_service.get_active_users("tenant_123")
    print(f"Active users: {active.active_users}")
    
    # Today metrics from cache
    today = await realtime_service.get_today_metrics("tenant_123")
    print(f"Today revenue: €{today.revenue}")
    print(f"Today orders: {today.orders}")
```

**Dettagli rapidi**:
- Senza Redis: passa `cache=None` per in-memory fallback (non persistente tra restart)
- Redis usa SET per active users, counters per revenue/orders
- TTL automatico: active users expire dopo 5 minuti di inattività

### 8. Router completo con auth

```python
from linkbay_analytics import create_analytics_router
from fastapi import Depends, HTTPException

async def get_current_tenant_id(token: str = Depends(oauth2_scheme)) -> str:
    # Tua logica di autenticazione
    user = await verify_token(token)
    return user.tenant_id

async def require_auth(token: str = Depends(oauth2_scheme)):
    if not await verify_token(token):
        raise HTTPException(status_code=401)

router = create_analytics_router(
    event_service=event_service,
    metrics_service=metrics_service,
    insights_service=insights_service,  # Opzionale
    benchmark_service=benchmark_service,  # Opzionale
    realtime_service=realtime_service,  # Opzionale
    get_current_tenant=get_current_tenant_id,  # Estrai tenant_id da auth
    dependencies=[Depends(require_auth)],  # Middleware auth globale
)

app.include_router(router, prefix="/api/v1/analytics")
```

Endpoint disponibili: `/events`, `/dashboard`, `/revenue/timeseries`, `/products/top`, `/insights` (se insights_service), `/benchmarks` (se benchmark_service), `/realtime/active-users`, `/realtime/today` (se realtime_service).

## Configurazione

### EventService

```python
EventService(
    event_model: Type[Any],  # Tuo modello SQLAlchemy o Django
    db_session: Any,  # Async session o connection
)
```

### MetricsService

```python
MetricsService(
    event_model: Type[Any],
    db_session: Any,
)
```

### InsightsService

```python
InsightsService(
    metrics_service: MetricsService,
    ai_service: Optional[AIServiceProtocol] = None,  # None = rule-based
)
```

### BenchmarkService

```python
BenchmarkService(
    event_model: Type[Any],
    db_session: Any,
    min_tenants: int = 5,  # Soglia privacy
)
```

### RealtimeService

```python
RealtimeService(
    cache: Optional[CacheProtocol] = None,  # None = in-memory
)
```

### create_analytics_router

```python
create_analytics_router(
    event_service: EventService,
    metrics_service: MetricsService,
    insights_service: Optional[InsightsService] = None,
    benchmark_service: Optional[BenchmarkService] = None,
    realtime_service: Optional[RealtimeService] = None,
    get_current_tenant: Optional[Callable] = None,
    dependencies: Optional[List[Depends]] = None,
)
```

**Override puntuali**: Tutti i servizi accettano parametri nel costruttore. Per override runtime (es. cambiare `period` per una query), passa parametri ai metodi del servizio: `get_dashboard_metrics(tenant_id, period=MetricsPeriod.CUSTOM, custom_from=..., custom_to=...)`.

## Troubleshooting

| Problema | Causa | Soluzione |
|----------|-------|-----------|
| `ImportError: No module named 'linkbay_analytics.models'` | v0.2.0 ha rimosso modelli built-in | Crea il tuo modello `AnalyticsEvent` seguendo schema in Quick Start |
| `AttributeError: 'NoneType' object has no attribute '__tablename__'` | `event_model` passato è None | Verifica che stai passando la classe del modello, non un'istanza |
| `asyncpg.exceptions.UndefinedTableError: relation "analytics_events" does not exist` | Tabella non creata nel database | Esegui migration Alembic o `CREATE TABLE analytics_events` manualmente |
| `redis.exceptions.ConnectionError: Error connecting to Redis` | Redis non disponibile ma `cache=redis_client` | Usa `cache=None` per in-memory fallback o avvia Redis |
| `HTTPException 400: Tenant ID required but get_current_tenant not configured` | `get_current_tenant` non fornito al router | Passa funzione che estrae tenant_id: `get_current_tenant=lambda: request.state.tenant_id` |
| Query lente su dashboard con molti eventi | Mancano indici su `tenant_id`, `created_at`, `event_name` | Crea indici: `CREATE INDEX idx_events_tenant ON analytics_events(tenant_id, created_at DESC)` |
| Insights sempre rule-based anche con `ai_service` | AI service non implementa `AIServiceProtocol` | Verifica che il tuo service abbia metodo `async def analyze_metrics(metrics, tenant_id)` |

## FAQ performance e sicurezza

- **Quanti eventi può gestire al secondo?**: Con PostgreSQL standard: ~1000 insert/s per core. Con TimescaleDB hypertables: ~10,000 insert/s. Per volumi maggiori, usa batch insert o queue (Celery + Redis).

- **Quanto spazio occupa 1M di eventi?**: ~200MB senza indici, ~400MB con indici su tenant_id, created_at, event_name. Il campo JSONB `properties` incide ~50 bytes/evento. Usa `pg_partman` o TimescaleDB per partitioning automatico.

- **I dati sono isolati tra tenant?**: Sì, tutte le query includono `WHERE tenant_id = :tenant_id`. Per sicurezza extra, usa PostgreSQL Row Level Security (RLS) con policy: `CREATE POLICY tenant_isolation ON analytics_events USING (tenant_id = current_setting('app.tenant_id'))`.

- **Redis è necessario?**: No. `RealtimeService(cache=None)` usa dizionario in-memory. Utile per dev/test, ma in produzione Redis è consigliato per active users persistenti tra restart e per scalare su più worker.

- **Come gestire GDPR/CCPA (cancellazione dati utente)?**: Implementa endpoint custom che esegue `DELETE FROM analytics_events WHERE user_id = :user_id AND tenant_id = :tenant_id`. Considera pseudonimizzazione: memorizza hash di user_id invece del valore originale.

- **Posso usare con Django ORM invece di SQLAlchemy?**: Sì. I servizi accettano qualsiasi `event_model` con attributi `__tablename__`, `id`, `tenant_id`, ecc. Passa Django model class e `db_session` può essere `connection.cursor()` per raw SQL.

- **Le aggregazioni SQL sono ottimizzate?**: Sì. Usiamo `DATE_TRUNC`, `PERCENTILE_CONT`, `COUNT DISTINCT`, `COALESCE` nativi PostgreSQL. No Pandas, no ORM overhead. Per dataset >10M eventi, considera materialized views con refresh periodico.

- **Benchmarks espongono dati sensibili?**: No. Usiamo percentili aggregati (p25, p50, p75) e richiediamo minimo 5 tenant (`min_tenants`) prima di mostrare statistiche. Nessun dato individuale è visibile.

## Licenza

MIT © Alessio Quagliara

## Supporto

- **Issues**: https://github.com/AlessioQuagliara/linkbay-analytics/issues
- **Email**: quagliara.alessio@gmail.com

Contribuisci, apri una issue o raccontaci come stai usando la libreria 🧡