chore: redefined and connected pricers

README.md

@@ -1,12 +1,5 @@
-<img width="200" align="left" src="https://github.com/user-attachments/assets/d148b00d-e9f9-4280-89cc-0cc866e17251" />
-
-### PHANTOM
-
 [![Build PDF](https://github.com/velocitatem/PHANTOM/actions/workflows/latex.yml/badge.svg)](https://github.com/velocitatem/PHANTOM/actions/workflows/latex.yml)
-[![TPU Research Cloud](https://img.shields.io/badge/TPU%20Research%20Cloud-TRC%20supported-4285F4?logo=googlecloud&logoColor=white)](https://sites.research.google/trc/faq/)
-[![Vercel Deploy](https://deploy-badge.vercel.app/?url=https://phantom-hotel.vercel.app&name=Hotel)](https://phantom-hotel.vercel.app)
-[![Vercel Deploy](https://deploy-badge.vercel.app/?url=https://phantom-airline.vercel.app&name=Airline)](https://phantom-airline.vercel.app)
-
-
 
+- https://phantom-hotel.vercel.app/
+- https://phantom-airline.vercel.app/
 

backend/provider/app.py

@@ -19,11 +19,11 @@ from procesing.pricers import (
     ElasticityBasedPricer
 )
 from procesing.steps import (
+    StateSpace,
     PredictPricesStep
 )
 from procesing import PipelineContext
 sys.path.append(os.path.dirname(os.path.abspath(__file__))+ "/../../lib/")
-print(os.path.dirname(os.path.abspath(__file__))+ "/../../lib/")
 from lib.model_registry import ModelRegistry
 
 # Config
@@ -53,12 +53,20 @@ def get_price(mode: Literal['hotel', 'airline'], productId: str, sessionId: Opti
     metadata = product['metadata']
     base_price = metadata.get('base_price', 100.0)
 
-    # fetch pre-computed prices from registry
-    prices_df = registry.get_prices('latest')
+    class Provider(SupabaseProvider, BackendAPIProvider):
+        def __init__(self, backend_url: str):
+            SupabaseProvider.__init__(self)
+            BackendAPIProvider.__init__(self, backend_url=backend_url)
+
+    context = PipelineContext(
+        provider=Provider(backend_url=os.getenv("BACKEND_URL")),
+        store_mode=mode
+    )
+
+    pricing_model = registry.get_pricing_model('latest')
     elasticity_df = registry.get_elasticity('latest')
 
-    if prices_df is None:
-        # fallback: no pre-computed prices available
+    if pricing_model is None or elasticity_df is None:
         return PriceResponse(
             productId=productId,
             price=base_price,
@@ -67,26 +75,87 @@ def get_price(mode: Literal['hotel', 'airline'], productId: str, sessionId: Opti
             elasticity=None
         )
 
-    # lookup pre-computed price for this product
+    products = context.products
+    if products.empty:
+        raise HTTPException(500, "No products available in catalog")
+
+    # merge elasticity with product base prices
+    products_with_meta = products.copy()
+    products_with_meta['base_price'] = products_with_meta['metadata'].apply(
+        lambda m: m.get('base_price', 100.0) if isinstance(m, dict) else 100.0
+    )
+
+    merged = products_with_meta[['id', 'base_price']].rename(
+        columns={'id': 'productId'}
+    ).merge(
+        elasticity_df[['productId', 'elasticity']],
+        on='productId',
+        how='left'
+    ).fillna({'elasticity': 0.0})
+
+    # compute demand: use pricer's mean_demand if available, else default
+    demand_values = (pricing_model.mean_demand
+                    if hasattr(pricing_model, 'mean_demand') and pricing_model.mean_demand is not None
+                    else np.ones(len(merged)) * 10.0)
+
+    # build state space with session features if sessionId provided
+    session_features = pd.DataFrame()
+    if sessionId:
+        try:
+            # fetch recent session interactions from backend
+            from procesing.steps.session import ExtractSessionFeaturesStep
+            import requests
+            from datetime import datetime, timedelta
+
+            t_end = datetime.utcnow()
+            t_start = t_end - timedelta(hours=1)
+            backend_url = os.getenv("BACKEND_URL")
+            print(backend_url)
+
+            resp = requests.get(
+                f"{os.getenv('BACKEND_URL')}/api/kafka/dump", # TODO: THIS IS SHIT, must fix this
+                params={'topic': 'user-interactions', 't_start': t_start.isoformat(), 't_end': t_end.isoformat()},
+                timeout=2
+            )
+
+            if resp.ok:
+                msgs = resp.json().get('messages', [])
+                interactions_df = pd.DataFrame(msgs)
+
+                if not interactions_df.empty and 'sessionId' in interactions_df.columns:
+                    session_interactions = interactions_df[interactions_df['sessionId'] == sessionId]
+
+                    if not session_interactions.empty:
+                        extractor = ExtractSessionFeaturesStep(context=context)
+                        session_features_df = extractor.transform(session_interactions)
+
+                        if not session_features_df.empty:
+                            session_features = session_features_df.drop(columns=['sessionId'])
+        except Exception as e:
+            print(f"[session-features-error] {e}")
+            # continue without session features
+
+    state = StateSpace(
+        demand=demand_values,
+        prices=merged['base_price'].values,
+        session_features=session_features,
+        product_ids=merged['productId'].values,
+        elasticity=merged['elasticity'].values,
+        metadata={'sessionId': sessionId, 'experimentId': experimentId}
+    )
+
+    oracle = PredictPricesStep(context=context)
+    prices_df = oracle.transform((pricing_model, state))
+
     product_price_row = prices_df[prices_df['productId'] == productId]
     if product_price_row.empty:
-        # product not in pre-computed prices, fallback to base
-        return PriceResponse(
-            productId=productId,
-            price=base_price,
-            base_price=base_price,
-            markup=1.0,
-            elasticity=None
-        )
+        raise HTTPException(404, f"No pricing available for product {productId}")
 
-    optimal_price = float(product_price_row['optimal_price'].iloc[0]) # TODO: use optimal_price everywhere as  aresult
+    optimal_price = float(product_price_row['predicted_price'].iloc[0])
 
-    # get elasticity if available
-    product_elasticity = None
-    if elasticity_df is not None:
-        product_elasticity_row = elasticity_df[elasticity_df['productId'] == productId]
-        if not product_elasticity_row.empty:
-            product_elasticity = float(product_elasticity_row['elasticity'].iloc[0])
+    product_elasticity_row = elasticity_df[elasticity_df['productId'] == productId]
+    product_elasticity = (float(product_elasticity_row['elasticity'].iloc[0])
+                         if not product_elasticity_row.empty else None)
 
     return PriceResponse(
         productId=productId,

backend/provider/requirements.txt

@@ -12,5 +12,4 @@ graphviz
 python-dotenv>=1.0.0
 requests>=2.31.0
 typing-extensions>=4.8.0
-pypickle
-pymc
+pickle5>=0.0.11; python_version < '3.8'

backend/server/app.py

@@ -290,7 +290,6 @@ async def get_products(
         query = supabase.table(table).select('*')
 
         # filter by exact date_index if provided
-        # dateIndex from frontend is days from today, convert to days since epoch
         if dateIndex is not None:
             query = query.eq('date_index', dateIndex)
 

docker-compose.e2e.yml

@@ -1,161 +0,0 @@
-# Docker Compose configuration for E2E testing
-# Usage: docker compose -f docker-compose.e2e.yml up -d
-#
-# This configuration runs only the services needed for E2E pricing tests:
-# - Backend API (event ingestion)
-# - Kafka + Zookeeper (event streaming)
-# - Redis (model registry)
-# - Pricing Provider (price serving)
-#
-# Excluded for E2E tests:
-# - Airflow (pipeline runs directly via test worker)
-# - PostgreSQL (not needed without Airflow)
-# - TensorBoard (ML visualization not needed)
-
-services:
-  # Backend API for event ingestion
-  backend:
-    container_name: "PHANTOM-e2e-backend"
-    build:
-      context: .
-      dockerfile: docker/backend.Dockerfile
-    ports:
-      - "${BACKEND_PORT:-5000}:5000"
-    environment:
-      - KAFKA_HOST=kafka
-      - KAFKA_PORT=29092
-      - BACKEND_PORT=5000
-      - NEXT_PUBLIC_SUPABASE_URL=${NEXT_PUBLIC_SUPABASE_URL}
-      - NEXT_PUBLIC_SUPABASE_ANON_KEY=${NEXT_PUBLIC_SUPABASE_ANON_KEY}
-    depends_on:
-      kafka:
-        condition: service_healthy
-    healthcheck:
-      test: ["CMD", "curl", "-f", "http://localhost:5000/health"]
-      interval: 10s
-      timeout: 5s
-      retries: 5
-      start_period: 10s
-    restart: unless-stopped
-
-  # Redis for model registry
-  redis:
-    container_name: "PHANTOM-e2e-redis"
-    build:
-      context: ./docker
-      dockerfile: Redis.dockerfile
-    ports:
-      - "${REDIS_PORT:-6378}:6379"
-    volumes:
-      - e2e_redis_data:/data
-    healthcheck:
-      test: ["CMD", "redis-cli", "ping"]
-      interval: 5s
-      timeout: 3s
-      retries: 5
-    restart: unless-stopped
-
-  # Zookeeper for Kafka coordination
-  zookeeper:
-    container_name: "PHANTOM-e2e-zookeeper"
-    build:
-      context: ./docker
-      dockerfile: Zookeeper.dockerfile
-    environment:
-      ZOOKEEPER_CLIENT_PORT: 2181
-    ports:
-      - "2181:2181"
-    healthcheck:
-      test: ["CMD-SHELL", "echo ruok | nc localhost 2181 | grep imok"]
-      interval: 10s
-      timeout: 5s
-      retries: 5
-    restart: unless-stopped
-
-  # Kafka for event streaming
-  kafka:
-    container_name: "PHANTOM-e2e-kafka"
-    build:
-      context: ./docker
-      dockerfile: Kafka.dockerfile
-    depends_on:
-      zookeeper:
-        condition: service_healthy
-    environment:
-      KAFKA_BROKER_ID: 1
-      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
-      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
-      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:29092,PLAINTEXT_HOST://localhost:9092
-      KAFKA_LISTENERS: PLAINTEXT://0.0.0.0:29092,PLAINTEXT_HOST://0.0.0.0:9092
-      KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
-      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
-      KAFKA_AUTO_CREATE_TOPICS_ENABLE: "true"
-      # Faster topic creation for tests
-      KAFKA_NUM_PARTITIONS: 1
-      KAFKA_DEFAULT_REPLICATION_FACTOR: 1
-    ports:
-      - "${KAFKA_PORT:-9092}:9092"
-    volumes:
-      - e2e_kafka_data:/var/lib/kafka/data
-    healthcheck:
-      test: ["CMD-SHELL", "kafka-topics.sh --bootstrap-server localhost:9092 --list"]
-      interval: 10s
-      timeout: 10s
-      retries: 10
-      start_period: 30s
-    restart: unless-stopped
-
-  # Redpanda Console for Kafka debugging (optional)
-  redpanda-console:
-    container_name: "PHANTOM-e2e-redpanda-console"
-    build:
-      context: ./docker
-      dockerfile: RedpandaConsole.dockerfile
-    depends_on:
-      kafka:
-        condition: service_healthy
-    environment:
-      KAFKA_BROKERS: kafka:29092
-    ports:
-      - "${REDPANDA_CONSOLE_PORT:-8080}:8080"
-    restart: unless-stopped
-    profiles:
-      - debug  # Only start with --profile debug
-
-  # Pricing Provider for serving prices
-  pricing-provider:
-    container_name: "PHANTOM-e2e-pricing-provider"
-    build:
-      context: .
-      dockerfile: docker/Provider.dockerfile
-    depends_on:
-      redis:
-        condition: service_healthy
-      kafka:
-        condition: service_healthy
-    environment:
-      - PROVIDER_PORT=5001
-      - REDIS_HOST=redis
-      - REDIS_PORT=6379
-      - KAFKA_HOST=kafka
-      - KAFKA_PORT=29092
-      - NEXT_PUBLIC_SUPABASE_URL=${NEXT_PUBLIC_SUPABASE_URL}
-      - NEXT_PUBLIC_SUPABASE_ANON_KEY=${NEXT_PUBLIC_SUPABASE_ANON_KEY}
-      - BACKEND_URL=http://backend:5000
-    ports:
-      - "${PROVIDER_PORT:-5001}:5001"
-    healthcheck:
-      test: ["CMD", "curl", "-f", "http://localhost:5001/health"]
-      interval: 10s
-      timeout: 5s
-      retries: 5
-      start_period: 10s
-    restart: unless-stopped
-
-volumes:
-  e2e_kafka_data:
-  e2e_redis_data:
-
-networks:
-  default:
-    name: phantom-e2e-network

docker-compose.yml

@@ -1,15 +1,4 @@
 services:
-
-  tensorboard:
-    image: tensorflow/tensorflow:latest
-    container_name: "PHANTOM-tensorboard"
-    ports:
-      - "6006:6006"
-    volumes:
-      - ./experiments/ml/runs:/logs
-    command: tensorboard --logdir=/logs --host=0.0.0.0 --port=6006
-    restart: unless-stopped
-
   backend:
     container_name: "PHANTOM-backend"
     build:
@@ -114,6 +103,12 @@ services:
       - _AIRFLOW_WWW_USER_PASSWORD=admin
       - REDIS_HOST=redis
       - REDIS_PORT=6379
+    volumes:
+      - ./experiments/airflow/dags:/opt/airflow/dags
+      - ./experiments/airflow/logs:/opt/airflow/logs
+      - ./experiments/airflow/plugins:/opt/airflow/plugins
+      - ./experiments/procesing:/opt/airflow/procesing
+      - ./lib:/opt/airflow/lib
     command: version
     restart: "no"
 
@@ -134,7 +129,6 @@ services:
       - AIRFLOW__CORE__LOAD_EXAMPLES=false
       - AIRFLOW__CORE__ENABLE_XCOM_PICKLING=true
       - AIRFLOW__WEBSERVER__EXPOSE_CONFIG=true
-      - AIRFLOW__WEBSERVER__SECRET_KEY=${AIRFLOW_SECRET_KEY}
       - KAFKA_HOST=kafka
       - KAFKA_PORT=29092
       - BACKEND_URL=http://backend:5000
@@ -144,6 +138,12 @@ services:
       - REDIS_PORT=6379
     ports:
       - "${AIRFLOW_WEBSERVER_PORT:-8085}:8080"
+    volumes:
+      - ./experiments/airflow/dags:/opt/airflow/dags:ro
+      - ./experiments/airflow/logs:/opt/airflow/logs
+      - ./experiments/airflow/plugins:/opt/airflow/plugins:ro
+      - ./experiments/procesing:/opt/airflow/procesing:ro
+      - ./lib:/opt/airflow/lib:ro
     command: webserver
     restart: unless-stopped
     healthcheck:
@@ -170,7 +170,6 @@ services:
       - AIRFLOW__CORE__DAGS_ARE_PAUSED_AT_CREATION=true
       - AIRFLOW__CORE__LOAD_EXAMPLES=false
       - AIRFLOW__CORE__ENABLE_XCOM_PICKLING=true
-      - AIRFLOW__WEBSERVER__SECRET_KEY=${AIRFLOW_SECRET_KEY}
       - KAFKA_HOST=kafka
       - KAFKA_PORT=29092
       - BACKEND_URL=http://backend:5000
@@ -178,6 +177,12 @@ services:
       - NEXT_PUBLIC_SUPABASE_ANON_KEY=${NEXT_PUBLIC_SUPABASE_ANON_KEY}
       - REDIS_HOST=redis
       - REDIS_PORT=6379
+    volumes:
+      - ./experiments/airflow/dags:/opt/airflow/dags:ro
+      - ./experiments/airflow/logs:/opt/airflow/logs
+      - ./experiments/airflow/plugins:/opt/airflow/plugins:ro
+      - ./experiments/procesing:/opt/airflow/procesing:ro
+      - ./lib:/opt/airflow/lib:ro
     command: scheduler
     restart: unless-stopped
     healthcheck:
@@ -203,9 +208,13 @@ services:
       - KAFKA_PORT=29092
       - NEXT_PUBLIC_SUPABASE_URL=${NEXT_PUBLIC_SUPABASE_URL}
       - NEXT_PUBLIC_SUPABASE_ANON_KEY=${NEXT_PUBLIC_SUPABASE_ANON_KEY}
-      - BACKEND_URL=http://localhost:5000
     ports:
       - "${PROVIDER_PORT:-5001}:5001"
+    volumes:
+      - ./lib:/app/lib:ro
+      - ./experiments/procesing:/app/procesing:ro
+      - ./backend/provider:/app/provider:ro
+    command: python -m uvicorn provider.app:app --host 0.0.0.0 --port 5001
     restart: unless-stopped
 
 volumes:

docker/Airflow.dockerfile

@@ -21,10 +21,3 @@ RUN pip install --no-cache-dir \
 
 # set airflow home
 ENV AIRFLOW_HOME=/opt/airflow
-
-COPY --chown=airflow:root experiments/airflow/dags ${AIRFLOW_HOME}/dags
-COPY --chown=airflow:root experiments/procesing ${AIRFLOW_HOME}/procesing
-COPY --chown=airflow:root lib ${AIRFLOW_HOME}/lib
-
-# create logs and plugins dirs (airflow expects them)
-RUN mkdir -p ${AIRFLOW_HOME}/logs ${AIRFLOW_HOME}/plugins

docker/Airflow.railway.dockerfile

@@ -1,41 +0,0 @@
-FROM apache/airflow:2.7.3-python3.11
-
-USER root
-
-RUN apt-get update && apt-get install -y --no-install-recommends \
-    build-essential \
-    supervisor \
-    && apt-get clean \
-    && rm -rf /var/lib/apt/lists/*
-
-USER airflow
-
-COPY requirements.txt /tmp/requirements.txt
-RUN pip install --no-cache-dir -r /tmp/requirements.txt
-
-RUN pip install --no-cache-dir \
-    psycopg2-binary \
-    apache-airflow-providers-postgres
-
-ENV AIRFLOW_HOME=/opt/airflow
-ENV AIRFLOW__CORE__EXECUTOR=SequentialExecutor
-ENV AIRFLOW__CORE__LOAD_EXAMPLES=false
-ENV AIRFLOW__CORE__ENABLE_XCOM_PICKLING=true
-ENV AIRFLOW__WEBSERVER__EXPOSE_CONFIG=true
-
-# copy all code into image (standalone - no volume mounts needed)
-COPY --chown=airflow:root experiments/airflow/dags ${AIRFLOW_HOME}/dags
-COPY --chown=airflow:root experiments/procesing ${AIRFLOW_HOME}/procesing
-COPY --chown=airflow:root lib ${AIRFLOW_HOME}/lib
-
-RUN mkdir -p ${AIRFLOW_HOME}/logs ${AIRFLOW_HOME}/plugins
-
-# copy entrypoint script
-COPY --chown=airflow:root docker/airflow-railway-entrypoint.sh /entrypoint.sh
-USER root
-RUN chmod +x /entrypoint.sh
-USER airflow
-
-EXPOSE 8080
-
-ENTRYPOINT ["/entrypoint.sh"]

docker/Provider.dockerfile

@@ -14,13 +14,11 @@ RUN apt-get update && apt-get install -y \
 COPY backend/provider/requirements.txt /app/
 RUN pip install --no-cache-dir -r requirements.txt
 
-# Copy application code into image
-COPY lib/ /app/lib/
-COPY experiments/procesing/ /app/procesing/
-COPY backend/provider/ /app/provider/
+# Structure will be mounted via volumes:
+# /app/lib -> lib/
+# /app/procesing -> experiments/procesing/
+# /app/provider -> backend/provider/
 
 ENV PYTHONPATH=/app:/app/lib:/app/procesing
 
-WORKDIR /app/provider
-
-CMD ["uvicorn", "app:app", "--host", "0.0.0.0", "--port", "5001"]
+CMD ["python", "-m", "uvicorn", "provider.app:app", "--host", "0.0.0.0", "--port", "5001"]

docker/airflow-railway-entrypoint.sh

@@ -1,20 +0,0 @@
-#!/bin/bash
-set -e
-
-# init db and create admin user on first run
-airflow db migrate
-
-# create admin user if not exists
-airflow users create \
-    --username "${AIRFLOW_ADMIN_USER:-admin}" \
-    --password "${AIRFLOW_ADMIN_PASSWORD:-admin}" \
-    --firstname Admin \
-    --lastname User \
-    --role Admin \
-    --email admin@example.com || true
-
-# start scheduler in background
-airflow scheduler &
-
-# start webserver in foreground (Railway needs one foreground process)
-exec airflow webserver --port ${PORT:-8080}

e2e/README.md

@@ -1,255 +0,0 @@
-# PHANTOM Dynamic Pricing E2E Test Suite
-
-End-to-end tests validating the dynamic pricing pipeline, including SimpleSurgePricer and SessionAwarePricer functionality.
-
-## System Under Test (SUT)
-
-```
-┌─────────────────────────────────────────────────────────────────────────┐
-│                        PHANTOM Pricing Pipeline                         │
-├─────────────────────────────────────────────────────────────────────────┤
-│                                                                         │
-│  ┌──────────────┐    ┌──────────────┐    ┌──────────────────────────┐  │
-│  │  Test Runner │───▶│ Backend API  │───▶│ Kafka (user-interactions)│  │
-│  │  (Playwright)│    │ POST /ingest │    │                          │  │
-│  └──────────────┘    └──────────────┘    └────────────┬─────────────┘  │
-│         │                                              │                │
-│         │                                              ▼                │
-│         │                                 ┌──────────────────────────┐  │
-│         │                                 │   Pipeline Worker        │  │
-│         │                                 │   - Fetch interactions   │  │
-│         │                                 │   - Compute demand       │  │
-│         │                                 │   - Apply surge pricing  │  │
-│         │                                 └────────────┬─────────────┘  │
-│         │                                              │                │
-│         │                                              ▼                │
-│         │                                 ┌──────────────────────────┐  │
-│         │                                 │   Redis (Model Registry) │  │
-│         │                                 │   - prices:latest        │  │
-│         │                                 └────────────┬─────────────┘  │
-│         │                                              │                │
-│         │                                              ▼                │
-│         │     ┌──────────────┐           ┌──────────────────────────┐  │
-│         └────▶│ Pricing API  │◀──────────│   Pricing Provider       │  │
-│               │ GET /price   │           │   (serves from Redis)    │  │
-│               └──────────────┘           └──────────────────────────┘  │
-│                                                                         │
-└─────────────────────────────────────────────────────────────────────────┘
-```
-
-## Test Scenarios
-
-| Scenario | Description | Expected Outcome |
-|----------|-------------|------------------|
-| **Baseline** | No interactions for product | Price = base_price (markup = 1.0) |
-| **Surge** | 5+ interactions (above threshold) | Price = base_price × 1.5 |
-| **Discount** | 1 interaction (at threshold) | Price = base_price × 0.9 |
-| **Multi-Product** | Different demand per product | Each product priced by its demand |
-| **Propagation** | Pipeline → Redis → API | Prices visible via API |
-| **Event Types** | Mix of view, click, cart | All events counted in demand |
-| **Multi-Session** | Events from different sessions | Demand aggregated correctly |
-
-## Test Configuration
-
-The tests use aggressive thresholds for fast feedback:
-
-```typescript
-pricing: {
-  highThreshold: 3,      // Surge after 3 interactions
-  lowThreshold: 1,       // Discount at ≤1 interaction
-  surgeMultiplier: 1.5,  // 50% price increase
-  discountMultiplier: 0.9, // 10% discount
-  windowSize: 10_000,    // 10 second window
-}
-```
-
-## Quick Start
-
-### 1. Start E2E Services
-
-```bash
-# Start minimal services for E2E testing
-docker compose -f docker-compose.e2e.yml up -d
-
-# Wait for services to be healthy
-docker compose -f docker-compose.e2e.yml ps
-
-# Optional: Start with Kafka UI for debugging
-docker compose -f docker-compose.e2e.yml --profile debug up -d
-```
-
-### 2. Install Test Dependencies
-
-```bash
-cd e2e
-npm install
-npx playwright install
-```
-
-### 3. Run Tests
-
-```bash
-# Run all E2E tests
-npm test
-
-# Run with UI (interactive mode)
-npm run test:ui
-
-# Run specific test file
-npm run test:pricing
-
-# Run in debug mode
-npm run test:debug
-
-# View test report
-npm run test:report
-```
-
-### 4. Cleanup
-
-```bash
-docker compose -f docker-compose.e2e.yml down -v
-```
-
-## Environment Variables
-
-| Variable | Default | Description |
-|----------|---------|-------------|
-| `BACKEND_URL` | `http://localhost:5000` | Backend API URL |
-| `PROVIDER_URL` | `http://localhost:5001` | Pricing Provider URL |
-| `REDIS_HOST` | `localhost` | Redis host |
-| `REDIS_PORT` | `6378` | Redis port |
-| `KAFKA_HOST` | `localhost` | Kafka host |
-| `KAFKA_PORT` | `9092` | Kafka port |
-
-## Test Architecture
-
-```
-e2e/
-├── playwright.config.ts    # Playwright configuration
-├── global-setup.ts         # Service health checks
-├── global-teardown.ts      # Cleanup
-├── package.json            # Dependencies and scripts
-├── tsconfig.json           # TypeScript configuration
-├── lib/
-│   ├── api-client.ts       # API interaction utilities
-│   ├── event-generator.ts  # Test event factory
-│   ├── pipeline-runner.ts  # TypeScript pipeline wrapper
-│   ├── pipeline-worker.py  # Python pipeline executor
-│   ├── fixtures.ts         # Playwright test fixtures
-│   └── index.ts            # Re-exports
-└── tests/
-    └── dynamic-pricing.spec.ts  # Main test file
-```
-
-## Pipeline Worker
-
-The tests use a dedicated Python pipeline worker (`lib/pipeline-worker.py`) instead of Airflow for faster, more reliable test execution.
-
-```bash
-# Run pipeline manually
-python3 lib/pipeline-worker.py \
-  --store-mode hotel \
-  --high-threshold 3 \
-  --surge-multiplier 1.5 \
-  --json-output
-
-# Dry run (no Redis publish)
-python3 lib/pipeline-worker.py --dry-run
-```
-
-## Debugging
-
-### View Kafka Events
-
-```bash
-# Via API
-curl "http://localhost:5000/api/kafka/dump?topic=user-interactions&last_n=10"
-
-# Via Redpanda Console (if started with --profile debug)
-open http://localhost:8080
-```
-
-### Check Redis State
-
-```bash
-docker exec -it PHANTOM-e2e-redis redis-cli
-> GET prices:latest
-> KEYS *
-```
-
-### View Pipeline Logs
-
-The pipeline worker logs detailed information:
-
-```
-[INFO] Starting E2E pricing pipeline: mode=hotel, high_threshold=3, surge_multiplier=1.5
-[INFO] Fetched 15 interaction records
-[INFO] Computed demand for 3 products
-[INFO] Applied surge pricing:
-  e2e-test...: base=$100.00 -> optimal=$150.00 (demand=5, markup=1.50x)
-[INFO] Published 3 prices to Redis
-```
-
-## Writing New Tests
-
-```typescript
-import { test, expect } from '../lib/fixtures';
-import { generateTestProductId } from '../lib/event-generator';
-
-test('my new pricing test', async ({ api, events, triggerPriceUpdate }) => {
-  // 1. Create unique product ID
-  const productId = generateTestProductId('my-test');
-
-  // 2. Log base price
-  await api.logPrice({
-    productId,
-    price: 100.0,
-    sessionId: events.session,
-    storeMode: 'hotel',
-  });
-
-  // 3. Generate events
-  const surgeEvents = events.generateSurgeSequence(productId, 5);
-  await api.ingestEvents(surgeEvents);
-
-  // 4. Trigger pipeline
-  const result = await triggerPriceUpdate();
-
-  // 5. Verify results
-  expect(result.success).toBe(true);
-  const pricedProduct = result.prices?.find(p => p.productId === productId);
-  expect(pricedProduct?.optimal_price).toBeGreaterThan(100);
-});
-```
-
-## Troubleshooting
-
-### "Backend not available"
-
-Ensure services are running:
-```bash
-docker compose -f docker-compose.e2e.yml ps
-docker compose -f docker-compose.e2e.yml logs backend
-```
-
-### "No interactions found"
-
-Check Kafka topic has events:
-```bash
-curl "http://localhost:5000/api/kafka/dump?topic=user-interactions"
-```
-
-### "Pipeline timeout"
-
-Increase timeout in `playwright.config.ts`:
-```typescript
-timeout: 180_000, // 3 minutes
-```
-
-### "Price not updated"
-
-Check Redis has latest prices:
-```bash
-docker exec -it PHANTOM-e2e-redis redis-cli GET prices:latest
-```

e2e/global-setup.ts

@@ -1,47 +0,0 @@
-import { testConfig } from './playwright.config';
-
-/**
- * Global setup for E2E tests
- * Verifies all services are healthy before running tests
- */
-async function globalSetup() {
-  console.log('\n🚀 PHANTOM E2E Test Suite - Global Setup\n');
-
-  // Check backend health
-  await checkService('Backend API', `${testConfig.backendUrl}/health`);
-
-  // Check pricing provider health
-  await checkService('Pricing Provider', `${testConfig.providerUrl}/health`);
-
-  console.log('\n✅ All services healthy. Starting tests...\n');
-}
-
-async function checkService(name: string, url: string): Promise<void> {
-  const maxRetries = 10;
-  const retryDelay = 2000;
-
-  for (let attempt = 1; attempt <= maxRetries; attempt++) {
-    try {
-      const response = await fetch(url);
-      if (response.ok) {
-        const data = await response.json();
-        console.log(`✅ ${name}: healthy`);
-        if (data.redis !== undefined) {
-          console.log(`   └─ Redis: ${data.redis ? 'connected' : 'disconnected'}`);
-        }
-        if (data.kafka !== undefined) {
-          console.log(`   └─ Kafka: ${data.kafka}`);
-        }
-        return;
-      }
-    } catch (error) {
-      if (attempt === maxRetries) {
-        throw new Error(`❌ ${name} is not available at ${url} after ${maxRetries} attempts`);
-      }
-      console.log(`⏳ Waiting for ${name} (attempt ${attempt}/${maxRetries})...`);
-      await new Promise(resolve => setTimeout(resolve, retryDelay));
-    }
-  }
-}
-
-export default globalSetup;

e2e/global-teardown.ts

@@ -1,10 +0,0 @@
-/**
- * Global teardown for E2E tests
- * Cleans up test data and resources
- */
-async function globalTeardown() {
-  console.log('\n🧹 PHANTOM E2E Test Suite - Global Teardown\n');
-  console.log('✅ Cleanup complete\n');
-}
-
-export default globalTeardown;

e2e/lib/api-client.ts

@@ -1,191 +0,0 @@
-import { testConfig } from '../playwright.config';
-
-/**
- * Event payload structure matching the backend API
- */
-export interface EventPayload {
-  sessionId: string;
-  experimentId?: string;
-  eventName: string;
-  page: string;
-  productId?: string;
-  metadata?: Record<string, unknown>;
-  storeMode: 'hotel' | 'airline';
-  userAgent?: string;
-  ts?: string;
-}
-
-/**
- * Price log payload structure
- */
-export interface PriceLogPayload {
-  productId: string;
-  price: number;
-  sessionId: string;
-  experimentId?: string;
-  storeMode: 'hotel' | 'airline';
-  ts?: string;
-}
-
-/**
- * Price response from the pricing provider
- */
-export interface PriceResponse {
-  productId: string;
-  price: number;
-  base_price: number;
-  markup: number;
-  elasticity: number | null;
-  model_version: string;
-}
-
-/**
- * API client for interacting with PHANTOM services
- */
-export class PhantomApiClient {
-  private backendUrl: string;
-  private providerUrl: string;
-
-  constructor(
-    backendUrl: string = testConfig.backendUrl,
-    providerUrl: string = testConfig.providerUrl
-  ) {
-    this.backendUrl = backendUrl;
-    this.providerUrl = providerUrl;
-  }
-
-  /**
-   * Send a user interaction event to the ingestion API
-   */
-  async ingestEvent(event: EventPayload): Promise<{ success: boolean }> {
-    const payload: EventPayload = {
-      ...event,
-      ts: event.ts || new Date().toISOString(),
-    };
-
-    const response = await fetch(`${this.backendUrl}/api/kafka/ingest`, {
-      method: 'POST',
-      headers: { 'Content-Type': 'application/json' },
-      body: JSON.stringify(payload),
-    });
-
-    if (!response.ok) {
-      throw new Error(`Failed to ingest event: ${response.status} ${await response.text()}`);
-    }
-
-    return response.json();
-  }
-
-  /**
-   * Send multiple events in rapid succession
-   */
-  async ingestEvents(events: EventPayload[], delayMs: number = 100): Promise<void> {
-    for (const event of events) {
-      await this.ingestEvent(event);
-      if (delayMs > 0) {
-        await new Promise(resolve => setTimeout(resolve, delayMs));
-      }
-    }
-  }
-
-  /**
-   * Log a price observation
-   */
-  async logPrice(priceLog: PriceLogPayload): Promise<{ success: boolean }> {
-    const payload: PriceLogPayload = {
-      ...priceLog,
-      ts: priceLog.ts || new Date().toISOString(),
-    };
-
-    const response = await fetch(`${this.backendUrl}/api/kafka/price-log`, {
-      method: 'POST',
-      headers: { 'Content-Type': 'application/json' },
-      body: JSON.stringify(payload),
-    });
-
-    if (!response.ok) {
-      throw new Error(`Failed to log price: ${response.status} ${await response.text()}`);
-    }
-
-    return response.json();
-  }
-
-  /**
-   * Get the current price for a product from the pricing provider
-   */
-  async getPrice(
-    mode: 'hotel' | 'airline',
-    productId: string,
-    sessionId?: string
-  ): Promise<PriceResponse> {
-    const params = new URLSearchParams();
-    if (sessionId) {
-      params.set('sessionId', sessionId);
-    }
-
-    const url = `${this.providerUrl}/api/${mode}/price/${productId}${params.toString() ? '?' + params.toString() : ''}`;
-    const response = await fetch(url);
-
-    if (!response.ok) {
-      throw new Error(`Failed to get price: ${response.status} ${await response.text()}`);
-    }
-
-    return response.json();
-  }
-
-  /**
-   * Dump events from Kafka topic for debugging
-   */
-  async dumpKafkaEvents(
-    topic: 'user-interactions' | 'price-logs' = 'user-interactions',
-    lastN?: number
-  ): Promise<{ success: boolean; count: number; data: unknown[] }> {
-    const params = new URLSearchParams({ topic });
-    if (lastN) {
-      params.set('last_n', String(lastN));
-    }
-
-    const response = await fetch(`${this.backendUrl}/api/kafka/dump?${params.toString()}`);
-
-    if (!response.ok) {
-      throw new Error(`Failed to dump Kafka events: ${response.status}`);
-    }
-
-    return response.json();
-  }
-
-  /**
-   * Check health of backend service
-   */
-  async checkBackendHealth(): Promise<{ status: string; kafka: string }> {
-    const response = await fetch(`${this.backendUrl}/health`);
-    return response.json();
-  }
-
-  /**
-   * Check health of pricing provider
-   */
-  async checkProviderHealth(): Promise<{ status: string; redis: boolean }> {
-    const response = await fetch(`${this.providerUrl}/health`);
-    return response.json();
-  }
-
-  /**
-   * List registered models in the pricing provider
-   */
-  async listModels(): Promise<Record<string, unknown>> {
-    const response = await fetch(`${this.providerUrl}/models`);
-    return response.json();
-  }
-
-  /**
-   * Reload models in the pricing provider
-   */
-  async reloadModels(): Promise<{ elasticity_loaded: boolean; pricing_model_loaded: boolean }> {
-    const response = await fetch(`${this.providerUrl}/models/reload`, { method: 'POST' });
-    return response.json();
-  }
-}
-
-// Singleton instance for convenience
-export const apiClient = new PhantomApiClient();

e2e/lib/event-generator.ts

@@ -1,249 +0,0 @@
-import { EventPayload, PriceLogPayload } from './api-client';
-import { v4 as uuidv4 } from 'uuid';
-
-/**
- * Canonical event names matching the frontend
- */
-export const EventNames = {
-  // Navigation events
-  PAGE_VIEW: 'page_view',
-  VIEW_ITEM_PAGE: 'view_item_page',
-  LEARN_MORE: 'learn_more_about_item',
-
-  // Cart events
-  ADD_TO_CART: 'add_item_to_cart',
-  REMOVE_FROM_CART: 'remove_item',
-  CHECKOUT_START: 'checkout_start',
-  PURCHASE_COMPLETE: 'purchase_complete',
-
-  // Search/Filter events
-  SEARCH: 'search',
-  FILTER_DATE: 'filter_for_date',
-  FILTER_AMENITIES: 'filter_for_amenities',
-  FILTER_PRICE: 'filter_for_price',
-  SORT_CHANGE: 'sort_change',
-
-  // Dwell signals (engagement)
-  HOVER_TITLE: 'hover_over_title',
-  HOVER_PARAGRAPH: 'hover_over_paragraph',
-  HOVER_LINK: 'hover_over_link',
-  HOVER_BUTTON: 'hover_over_button',
-
-  // Session
-  SESSION_START: 'session_start',
-} as const;
-
-export type EventName = typeof EventNames[keyof typeof EventNames];
-
-/**
- * Test product configuration
- */
-export interface TestProduct {
-  id: string;
-  basePrice: number;
-  storeMode: 'hotel' | 'airline';
-  name?: string;
-}
-
-/**
- * Generates test events for dynamic pricing E2E tests
- */
-export class EventGenerator {
-  private sessionId: string;
-  private experimentId: string;
-  private storeMode: 'hotel' | 'airline';
-
-  constructor(options?: {
-    sessionId?: string;
-    experimentId?: string;
-    storeMode?: 'hotel' | 'airline';
-  }) {
-    this.sessionId = options?.sessionId || uuidv4();
-    this.experimentId = options?.experimentId || uuidv4();
-    this.storeMode = options?.storeMode || 'hotel';
-  }
-
-  get session(): string {
-    return this.sessionId;
-  }
-
-  get experiment(): string {
-    return this.experimentId;
-  }
-
-  /**
-   * Create a new session for isolation between test scenarios
-   */
-  newSession(): string {
-    this.sessionId = uuidv4();
-    return this.sessionId;
-  }
-
-  /**
-   * Generate a single event
-   */
-  createEvent(
-    eventName: EventName,
-    productId: string,
-    metadata?: Record<string, unknown>
-  ): EventPayload {
-    return {
-      sessionId: this.sessionId,
-      experimentId: this.experimentId,
-      eventName,
-      page: `/${this.storeMode}/products/${productId}`,
-      productId,
-      metadata: metadata || {},
-      storeMode: this.storeMode,
-      userAgent: 'PHANTOM-E2E-Test/1.0',
-      ts: new Date().toISOString(),
-    };
-  }
-
-  /**
-   * Generate a product view event
-   */
-  viewProduct(productId: string): EventPayload {
-    return this.createEvent(EventNames.VIEW_ITEM_PAGE, productId, {
-      referrer: `/${this.storeMode}/products`,
-      viewport: { width: 1920, height: 1080 },
-    });
-  }
-
-  /**
-   * Generate a "learn more" event (high intent signal)
-   */
-  learnMore(productId: string): EventPayload {
-    return this.createEvent(EventNames.LEARN_MORE, productId, {
-      section: 'details',
-    });
-  }
-
-  /**
-   * Generate a hover event (engagement signal)
-   */
-  hover(productId: string, element: 'title' | 'paragraph' | 'button' = 'title'): EventPayload {
-    const eventMap = {
-      title: EventNames.HOVER_TITLE,
-      paragraph: EventNames.HOVER_PARAGRAPH,
-      button: EventNames.HOVER_BUTTON,
-    };
-    return this.createEvent(eventMap[element], productId, {
-      duration_ms: Math.floor(Math.random() * 2000) + 500,
-    });
-  }
-
-  /**
-   * Generate an add-to-cart event
-   */
-  addToCart(productId: string, quantity: number = 1): EventPayload {
-    return this.createEvent(EventNames.ADD_TO_CART, productId, {
-      quantity,
-      cart_size: quantity,
-    });
-  }
-
-  /**
-   * Generate a sequence of high-velocity events for surge pricing trigger
-   * This simulates rapid user interest in a product
-   */
-  generateSurgeSequence(productId: string, count: number): EventPayload[] {
-    const events: EventPayload[] = [];
-
-    for (let i = 0; i < count; i++) {
-      // Mix of different event types to simulate realistic behavior
-      events.push(this.viewProduct(productId));
-
-      if (i % 2 === 0) {
-        events.push(this.learnMore(productId));
-      }
-
-      if (i % 3 === 0) {
-        events.push(this.hover(productId, 'title'));
-      }
-    }
-
-    return events;
-  }
-
-  /**
-   * Generate a normal browsing session (not triggering surge)
-   */
-  generateNormalSession(productId: string): EventPayload[] {
-    return [
-      this.viewProduct(productId),
-      this.hover(productId, 'title'),
-    ];
-  }
-
-  /**
-   * Generate high-velocity agent-like behavior
-   * This should trigger SessionAwarePricer's agent detection
-   */
-  generateAgentBehavior(productIds: string[]): EventPayload[] {
-    const events: EventPayload[] = [];
-
-    // Rapid-fire product views across multiple products
-    for (const productId of productIds) {
-      events.push(this.viewProduct(productId));
-      // Very quick succession - agent-like behavior
-    }
-
-    return events;
-  }
-
-  /**
-   * Generate a price log entry
-   */
-  createPriceLog(productId: string, price: number): PriceLogPayload {
-    return {
-      productId,
-      price,
-      sessionId: this.sessionId,
-      experimentId: this.experimentId,
-      storeMode: this.storeMode,
-      ts: new Date().toISOString(),
-    };
-  }
-}
-
-/**
- * Pre-configured test products for E2E tests
- * These should match products in your test database
- */
-export const TestProducts = {
-  // Hotel products with known base prices
-  hotel1: {
-    id: 'e2e-test-hotel-001',
-    basePrice: 150.00,
-    storeMode: 'hotel' as const,
-    name: 'E2E Test Hotel 1',
-  },
-  hotel2: {
-    id: 'e2e-test-hotel-002',
-    basePrice: 200.00,
-    storeMode: 'hotel' as const,
-    name: 'E2E Test Hotel 2',
-  },
-  hotel3: {
-    id: 'e2e-test-hotel-003',
-    basePrice: 100.00,
-    storeMode: 'hotel' as const,
-    name: 'E2E Test Hotel 3',
-  },
-
-  // Airline products
-  airline1: {
-    id: 'e2e-test-airline-001',
-    basePrice: 350.00,
-    storeMode: 'airline' as const,
-    name: 'E2E Test Flight 1',
-  },
-};
-
-/**
- * Generate a unique test product ID for isolation
- */
-export function generateTestProductId(prefix: string = 'e2e-test'): string {
-  return `${prefix}-${uuidv4().slice(0, 8)}`;
-}

e2e/lib/fixtures.ts

@@ -1,143 +0,0 @@
-import { test as base, expect } from '@playwright/test';
-import { PhantomApiClient, apiClient } from './api-client';
-import { EventGenerator, TestProducts } from './event-generator';
-import { runPricingPipeline, waitForPriceUpdate, PipelineResult } from './pipeline-runner';
-import { testConfig } from '../playwright.config';
-
-/**
- * Extended test fixtures for PHANTOM E2E tests
- */
-export interface PhantomTestFixtures {
-  /** API client for interacting with PHANTOM services */
-  api: PhantomApiClient;
-
-  /** Event generator for creating test events */
-  events: EventGenerator;
-
-  /** Run the pricing pipeline and wait for updates */
-  triggerPriceUpdate: (options?: {
-    storeMode?: 'hotel' | 'airline';
-    highThreshold?: number;
-    lowThreshold?: number;
-    surgeMultiplier?: number;
-    discountMultiplier?: number;
-  }) => Promise<PipelineResult>;
-
-  /** Wait for a specific price condition */
-  waitForPrice: (
-    productId: string,
-    condition: (price: number, basePrice: number) => boolean,
-    storeMode?: 'hotel' | 'airline'
-  ) => Promise<{ price: number; basePrice: number; markup: number }>;
-
-  /** Test configuration */
-  config: typeof testConfig;
-}
-
-/**
- * Custom test with PHANTOM fixtures
- */
-export const test = base.extend<PhantomTestFixtures>({
-  api: async ({}, use) => {
-    await use(apiClient);
-  },
-
-  events: async ({}, use) => {
-    // Create a new event generator with a fresh session for each test
-    const generator = new EventGenerator({
-      storeMode: 'hotel',
-    });
-    await use(generator);
-  },
-
-  triggerPriceUpdate: async ({}, use) => {
-    const trigger = async (options = {}) => {
-      const result = await runPricingPipeline({
-        storeMode: 'hotel',
-        highThreshold: testConfig.pricing.highThreshold,
-        lowThreshold: testConfig.pricing.lowThreshold,
-        surgeMultiplier: testConfig.pricing.surgeMultiplier,
-        discountMultiplier: testConfig.pricing.discountMultiplier,
-        ...options,
-      });
-
-      // Wait a moment for Redis to be fully updated
-      await new Promise(resolve => setTimeout(resolve, 500));
-
-      return result;
-    };
-
-    await use(trigger);
-  },
-
-  waitForPrice: async ({ api }, use) => {
-    const waiter = async (
-      productId: string,
-      condition: (price: number, basePrice: number) => boolean,
-      storeMode: 'hotel' | 'airline' = 'hotel'
-    ) => {
-      let lastPrice = 0;
-      let lastBasePrice = 0;
-
-      const updated = await waitForPriceUpdate(async () => {
-        const priceResponse = await api.getPrice(storeMode, productId);
-        lastPrice = priceResponse.price;
-        lastBasePrice = priceResponse.base_price;
-        return condition(priceResponse.price, priceResponse.base_price);
-      });
-
-      if (!updated) {
-        throw new Error(
-          `Price condition not met within timeout. Last price: ${lastPrice}, base: ${lastBasePrice}`
-        );
-      }
-
-      return {
-        price: lastPrice,
-        basePrice: lastBasePrice,
-        markup: lastPrice / lastBasePrice,
-      };
-    };
-
-    await use(waiter);
-  },
-
-  config: async ({}, use) => {
-    await use(testConfig);
-  },
-});
-
-export { expect };
-
-/**
- * Helper assertions for pricing tests
- */
-export const PricingAssertions = {
-  /**
-   * Assert that a price has surge markup applied
-   */
-  isSurged: (price: number, basePrice: number, expectedMultiplier: number, tolerance = 0.01) => {
-    const actualMarkup = price / basePrice;
-    const minExpected = expectedMultiplier * (1 - tolerance);
-    const maxExpected = expectedMultiplier * (1 + tolerance);
-    return actualMarkup >= minExpected && actualMarkup <= maxExpected;
-  },
-
-  /**
-   * Assert that a price has discount applied
-   */
-  isDiscounted: (price: number, basePrice: number, expectedMultiplier: number, tolerance = 0.01) => {
-    const actualMarkup = price / basePrice;
-    const minExpected = expectedMultiplier * (1 - tolerance);
-    const maxExpected = expectedMultiplier * (1 + tolerance);
-    return actualMarkup >= minExpected && actualMarkup <= maxExpected;
-  },
-
-  /**
-   * Assert that a price is at base (no surge/discount)
-   */
-  isBase: (price: number, basePrice: number, tolerance = 0.01) => {
-    const actualMarkup = price / basePrice;
-    return actualMarkup >= (1 - tolerance) && actualMarkup <= (1 + tolerance);
-  },
-};

e2e/lib/index.ts

@@ -1,6 +0,0 @@
-// Re-export all test utilities
-
-export * from './api-client';
-export * from './event-generator';
-export * from './pipeline-runner';
-export * from './fixtures';

e2e/lib/pipeline-runner.ts

@@ -1,152 +0,0 @@
-import { spawn } from 'child_process';
-import path from 'path';
-import { testConfig } from '../playwright.config';
-
-/**
- * Pipeline execution result
- */
-export interface PipelineResult {
-  success: boolean;
-  interactions_count: number;
-  products_count: number;
-  prices_published: boolean;
-  prices?: Array<{
-    productId: string;
-    current_price: number;
-    base_price: number;
-    optimal_price: number;
-    demand_score: number;
-  }>;
-  timestamp?: string;
-  message?: string;
-  error?: string;
-}
-
-/**
- * Pipeline configuration options
- */
-export interface PipelineOptions {
-  storeMode?: 'hotel' | 'airline';
-  highThreshold?: number;
-  lowThreshold?: number;
-  surgeMultiplier?: number;
-  discountMultiplier?: number;
-  dryRun?: boolean;
-}
-
-/**
- * Execute the pricing pipeline to update prices based on current events
- */
-export async function runPricingPipeline(options: PipelineOptions = {}): Promise<PipelineResult> {
-  const {
-    storeMode = 'hotel',
-    highThreshold = testConfig.pricing.highThreshold,
-    lowThreshold = testConfig.pricing.lowThreshold,
-    surgeMultiplier = testConfig.pricing.surgeMultiplier,
-    discountMultiplier = testConfig.pricing.discountMultiplier,
-    dryRun = false,
-  } = options;
-
-  const workerPath = path.join(__dirname, 'pipeline-worker.py');
-
-  const args = [
-    workerPath,
-    '--store-mode', storeMode,
-    '--high-threshold', String(highThreshold),
-    '--low-threshold', String(lowThreshold),
-    '--surge-multiplier', String(surgeMultiplier),
-    '--discount-multiplier', String(discountMultiplier),
-    '--json-output',
-  ];
-
-  if (dryRun) {
-    args.push('--dry-run');
-  }
-
-  return new Promise((resolve, reject) => {
-    const python = spawn('python3', args, {
-      env: {
-        ...process.env,
-        BACKEND_URL: testConfig.backendUrl,
-        REDIS_HOST: testConfig.redisHost,
-        REDIS_PORT: String(testConfig.redisPort),
-        KAFKA_HOST: testConfig.kafkaHost,
-        KAFKA_PORT: String(testConfig.kafkaPort),
-      },
-    });
-
-    let stdout = '';
-    let stderr = '';
-
-    python.stdout.on('data', (data) => {
-      stdout += data.toString();
-    });
-
-    python.stderr.on('data', (data) => {
-      stderr += data.toString();
-      // Log pipeline output for debugging
-      console.log('[Pipeline]', data.toString().trim());
-    });
-
-    python.on('close', (code) => {
-      if (code === 0) {
-        try {
-          // Find JSON output in stdout (last JSON object)
-          const jsonMatch = stdout.match(/\{[\s\S]*\}$/);
-          if (jsonMatch) {
-            const result = JSON.parse(jsonMatch[0]);
-            resolve(result);
-          } else {
-            resolve({
-              success: true,
-              interactions_count: 0,
-              products_count: 0,
-              prices_published: false,
-              message: 'Pipeline completed but no JSON output',
-            });
-          }
-        } catch (parseError) {
-          resolve({
-            success: true,
-            interactions_count: 0,
-            products_count: 0,
-            prices_published: false,
-            message: 'Pipeline completed but output not parseable',
-          });
-        }
-      } else {
-        reject(new Error(`Pipeline exited with code ${code}: ${stderr}`));
-      }
-    });
-
-    python.on('error', (error) => {
-      reject(new Error(`Failed to start pipeline: ${error.message}`));
-    });
-  });
-}
-
-/**
- * Wait for prices to be updated in Redis and available via the pricing API
- */
-export async function waitForPriceUpdate(
-  checkFn: () => Promise<boolean>,
-  maxWaitMs: number = testConfig.timing.maxPriceWait,
-  intervalMs: number = testConfig.timing.priceCheckInterval
-): Promise<boolean> {
-  const startTime = Date.now();
-
-  while (Date.now() - startTime < maxWaitMs) {
-    try {
-      const updated = await checkFn();
-      if (updated) {
-        return true;
-      }
-    } catch (error) {
-      // Ignore errors during polling
-    }
-
-    await new Promise(resolve => setTimeout(resolve, intervalMs));
-  }
-
-  return false;
-}

e2e/lib/pipeline-worker.py

@@ -1,245 +0,0 @@
-#!/usr/bin/env python3
-"""
-E2E Test Pipeline Worker
-
-A lightweight worker that runs the surge pricing pipeline for E2E tests.
-This bypasses Airflow for faster, more reliable test execution.
-
-Usage:
-    python pipeline-worker.py --store-mode hotel --high-threshold 3 --surge-multiplier 1.5
-"""
-
-import argparse
-import json
-import logging
-import os
-import sys
-from typing import Optional
-from datetime import datetime
-
-# Add project paths
-project_root = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
-sys.path.insert(0, project_root)
-sys.path.insert(0, os.path.join(project_root, 'experiments'))
-sys.path.insert(0, os.path.join(project_root, 'lib'))
-
-from procesing.context import PipelineContext
-from procesing.providers import BackendAPIProvider
-from procesing.steps import (
-    FetchInteractionsStep,
-    FetchPriceLogsStep,
-    ComputeDemandStep,
-    AggregatePriceLogsStep,
-    JoinProductFeaturesStep,
-)
-from procesing.pricers.simple import SimpleSurgePricer
-from lib.model_registry import ModelRegistry
-
-logging.basicConfig(
-    level=logging.INFO,
-    format='%(asctime)s [%(levelname)s] %(message)s'
-)
-log = logging.getLogger(__name__)
-
-
-class E2ETestProvider(BackendAPIProvider):
-    """Provider configured for E2E test environment"""
-
-    def __init__(self, backend_url: str = None):
-        self.backend_url = backend_url or os.getenv('BACKEND_URL', 'http://localhost:5000')
-        super().__init__()
-
-
-def run_pricing_pipeline(
-    store_mode: str = 'hotel',
-    high_threshold: int = 3,
-    low_threshold: int = 1,
-    surge_multiplier: float = 1.5,
-    discount_multiplier: float = 0.9,
-    dry_run: bool = False
-) -> dict:
-    """
-    Execute the surge pricing pipeline and publish results to Redis.
-
-    Args:
-        store_mode: 'hotel' or 'airline'
-        high_threshold: Demand threshold for surge pricing
-        low_threshold: Demand threshold for discount pricing
-        surge_multiplier: Price multiplier for high demand
-        discount_multiplier: Price multiplier for low demand
-        dry_run: If True, don't publish to Redis
-
-    Returns:
-        dict with pipeline results and statistics
-    """
-    log.info(f"Starting E2E pricing pipeline: mode={store_mode}, "
-             f"high_threshold={high_threshold}, surge_multiplier={surge_multiplier}")
-
-    # Initialize provider and context
-    provider = E2ETestProvider()
-    context = PipelineContext(provider=provider, store_mode=store_mode)
-
-    # Step 1: Fetch interactions from Kafka
-    log.info("Fetching interactions from Kafka...")
-    fetch_interactions = FetchInteractionsStep(context)
-    interactions_df = fetch_interactions.transform(None)
-    log.info(f"Fetched {len(interactions_df)} interaction records")
-
-    if interactions_df.empty:
-        log.warning("No interactions found. Pipeline will produce no price updates.")
-        return {
-            'success': True,
-            'interactions_count': 0,
-            'products_count': 0,
-            'prices_published': False,
-            'message': 'No interactions to process'
-        }
-
-    # Step 2: Fetch price logs from Kafka
-    log.info("Fetching price logs from Kafka...")
-    fetch_prices = FetchPriceLogsStep(context)
-    price_logs_df = fetch_prices.transform(None)
-    log.info(f"Fetched {len(price_logs_df)} price log records")
-
-    # Step 3: Compute demand scores
-    log.info("Computing demand scores...")
-    compute_demand = ComputeDemandStep(context)
-    demand_df = compute_demand.transform(interactions_df)
-    log.info(f"Computed demand for {len(demand_df)} products")
-
-    if demand_df.empty:
-        log.warning("No demand data computed.")
-        return {
-            'success': True,
-            'interactions_count': len(interactions_df),
-            'products_count': 0,
-            'prices_published': False,
-            'message': 'No demand data to process'
-        }
-
-    # Step 4: Aggregate price logs
-    log.info("Aggregating price logs...")
-    aggregate_prices = AggregatePriceLogsStep(context)
-    price_agg_df = aggregate_prices.transform(price_logs_df)
-    log.info(f"Aggregated prices for {len(price_agg_df)} products")
-
-    # Step 5: Join product features
-    log.info("Joining product features...")
-    join_features = JoinProductFeaturesStep(context)
-    features_df = join_features.transform((demand_df, price_agg_df))
-    log.info(f"Joined features for {len(features_df)} products")
-
-    if features_df.empty:
-        log.warning("No product features after join.")
-        return {
-            'success': True,
-            'interactions_count': len(interactions_df),
-            'products_count': 0,
-            'prices_published': False,
-            'message': 'No product features to price'
-        }
-
-    # Step 6: Apply surge pricing
-    log.info(f"Applying surge pricing (high={high_threshold}, surge={surge_multiplier}x)...")
-
-    # Rename columns for pricer compatibility
-    data = features_df.rename(columns={'demand_score': 'demand'})
-
-    surge_pricer = SimpleSurgePricer(
-        high_threshold=high_threshold,
-        low_threshold=low_threshold,
-        surge_multiplier=surge_multiplier,
-        discount_multiplier=discount_multiplier
-    )
-    surge_pricer.fit(data)
-    data['optimal_price'] = surge_pricer.predict()
-
-    # Prepare output DataFrame
-    prices_df = data[['productId', 'price', 'base_price', 'optimal_price', 'demand']].rename(columns={
-        'price': 'current_price',
-        'demand': 'demand_score'
-    })
-
-    log.info(f"Generated optimal prices for {len(prices_df)} products")
-
-    # Log pricing decisions
-    for _, row in prices_df.iterrows():
-        markup = row['optimal_price'] / row['base_price'] if row['base_price'] > 0 else 1.0
-        log.info(f"  {row['productId'][:8]}...: base=${row['base_price']:.2f} "
-                 f"-> optimal=${row['optimal_price']:.2f} (demand={row['demand_score']:.0f}, markup={markup:.2f}x)")
-
-    # Step 7: Publish to Redis
-    if not dry_run:
-        log.info("Publishing prices to Redis registry...")
-        registry = ModelRegistry()
-
-        metadata = {
-            'timestamp': datetime.utcnow().isoformat(),
-            'store_mode': store_mode,
-            'pipeline': 'e2e_test_worker',
-            'high_threshold': high_threshold,
-            'low_threshold': low_threshold,
-            'surge_multiplier': surge_multiplier,
-            'discount_multiplier': discount_multiplier,
-        }
-
-        registry.publish_prices(prices_df, model_name='latest', metadata=metadata)
-        log.info(f"✅ Published {len(prices_df)} prices to Redis")
-    else:
-        log.info("Dry run - skipping Redis publish")
-
-    return {
-        'success': True,
-        'interactions_count': len(interactions_df),
-        'products_count': len(prices_df),
-        'prices_published': not dry_run,
-        'prices': prices_df.to_dict(orient='records'),
-        'timestamp': datetime.utcnow().isoformat()
-    }
-
-
-def main():
-    parser = argparse.ArgumentParser(description='E2E Test Pipeline Worker')
-    parser.add_argument('--store-mode', choices=['hotel', 'airline'], default='hotel',
-                        help='Store mode (hotel or airline)')
-    parser.add_argument('--high-threshold', type=int, default=3,
-                        help='Demand threshold for surge pricing')
-    parser.add_argument('--low-threshold', type=int, default=1,
-                        help='Demand threshold for discount pricing')
-    parser.add_argument('--surge-multiplier', type=float, default=1.5,
-                        help='Price multiplier for high demand')
-    parser.add_argument('--discount-multiplier', type=float, default=0.9,
-                        help='Price multiplier for low demand')
-    parser.add_argument('--dry-run', action='store_true',
-                        help='Run without publishing to Redis')
-    parser.add_argument('--json-output', action='store_true',
-                        help='Output results as JSON')
-
-    args = parser.parse_args()
-
-    try:
-        result = run_pricing_pipeline(
-            store_mode=args.store_mode,
-            high_threshold=args.high_threshold,
-            low_threshold=args.low_threshold,
-            surge_multiplier=args.surge_multiplier,
-            discount_multiplier=args.discount_multiplier,
-            dry_run=args.dry_run
-        )
-
-        if args.json_output:
-            print(json.dumps(result, indent=2))
-        else:
-            log.info(f"Pipeline completed: {result['products_count']} products priced")
-
-        sys.exit(0 if result['success'] else 1)
-
-    except Exception as e:
-        log.error(f"Pipeline failed: {e}")
-        if args.json_output:
-            print(json.dumps({'success': False, 'error': str(e)}))
-        sys.exit(1)
-
-
-if __name__ == '__main__':
-    main()

e2e/package.json

@@ -1,27 +0,0 @@
-{
-  "name": "phantom-e2e-tests",
-  "version": "1.0.0",
-  "description": "E2E tests for PHANTOM Dynamic Pricing Pipeline",
-  "scripts": {
-    "test": "playwright test",
-    "test:ui": "playwright test --ui",
-    "test:headed": "playwright test --headed",
-    "test:debug": "playwright test --debug",
-    "test:report": "playwright show-report",
-    "test:pricing": "playwright test dynamic-pricing",
-    "test:health": "playwright test --grep 'health'",
-    "pipeline:run": "python3 lib/pipeline-worker.py --store-mode hotel --high-threshold 3 --surge-multiplier 1.5",
-    "pipeline:dry-run": "python3 lib/pipeline-worker.py --dry-run --json-output",
-    "services:check": "curl -s http://localhost:5000/health && curl -s http://localhost:5001/health"
-  },
-  "devDependencies": {
-    "@playwright/test": "^1.49.0",
-    "@types/node": "^20.0.0",
-    "typescript": "^5.0.0",
-    "uuid": "^9.0.0",
-    "@types/uuid": "^9.0.0"
-  },
-  "engines": {
-    "node": ">=18.0.0"
-  }
-}

e2e/playwright.config.ts

@@ -1,84 +0,0 @@
-import { defineConfig, devices } from '@playwright/test';
-
-/**
- * Playwright configuration for PHANTOM Dynamic Pricing E2E Tests
- *
- * Tests validate the entire pricing pipeline:
- * Frontend Events → Kafka → Pipeline Processing → Redis → Pricing API
- */
-export default defineConfig({
-  testDir: './tests',
-  fullyParallel: false, // Run tests sequentially to avoid race conditions in shared state
-  forbidOnly: !!process.env.CI,
-  retries: process.env.CI ? 2 : 0,
-  workers: 1, // Single worker for E2E tests to ensure isolation
-  reporter: [
-    ['html', { outputFolder: 'playwright-report' }],
-    ['list']
-  ],
-
-  // Global timeout for each test
-  timeout: 120_000, // 2 minutes per test (includes pipeline processing time)
-
-  // Expect timeout for assertions
-  expect: {
-    timeout: 30_000, // 30 seconds for price updates to propagate
-  },
-
-  use: {
-    // Base URL for the backend API
-    baseURL: process.env.BACKEND_URL || 'http://localhost:5000',
-
-    // Collect trace on first retry
-    trace: 'on-first-retry',
-
-    // Screenshot on failure
-    screenshot: 'only-on-failure',
-  },
-
-  // Global setup and teardown
-  globalSetup: require.resolve('./global-setup'),
-  globalTeardown: require.resolve('./global-teardown'),
-
-  projects: [
-    {
-      name: 'dynamic-pricing',
-      testMatch: /.*\.spec\.ts/,
-    },
-  ],
-
-  // Environment configuration
-  // These can be overridden via environment variables
-});
-
-// Export test configuration constants
-export const testConfig = {
-  // API endpoints
-  backendUrl: process.env.BACKEND_URL || 'http://localhost:5000',
-  providerUrl: process.env.PROVIDER_URL || 'http://localhost:5001',
-
-  // Redis configuration
-  redisHost: process.env.REDIS_HOST || 'localhost',
-  redisPort: parseInt(process.env.REDIS_PORT || '6378'),
-
-  // Kafka configuration
-  kafkaHost: process.env.KAFKA_HOST || 'localhost',
-  kafkaPort: parseInt(process.env.KAFKA_PORT || '9092'),
-
-  // Pricing thresholds for tests (aggressive settings for fast feedback)
-  pricing: {
-    highThreshold: 3,     // Trigger surge after 3 interactions
-    lowThreshold: 1,      // Trigger discount at 1 or fewer interactions
-    surgeMultiplier: 1.5, // 50% price increase on surge
-    discountMultiplier: 0.9, // 10% discount on low demand
-    windowSize: 10_000,   // 10 second window for demand calculation
-  },
-
-  // Timing configuration
-  timing: {
-    eventDelay: 100,           // Delay between events (ms)
-    pipelineWait: 5_000,       // Wait for pipeline processing (ms)
-    priceCheckInterval: 1_000, // Interval between price checks (ms)
-    maxPriceWait: 30_000,      // Max wait for price update (ms)
-  },
-};

e2e/tests/dynamic-pricing.spec.ts

@@ -1,497 +0,0 @@
-/**
- * PHANTOM Dynamic Pricing E2E Test Suite
- *
- * Validates that SimpleSurgePricer and SessionAwarePricer correctly adjust
- * product prices in real-time based on high-velocity user interactions.
- *
- * System Under Test (SUT):
- * - Frontend (interaction generation via API calls)
- * - Backend API (POST /api/ingest → Kafka)
- * - Kafka (user-interactions topic)
- * - Pipeline Worker (demand calculation → surge pricing)
- * - Redis (model registry)
- * - Pricing Provider (GET /api/{mode}/price/{productId})
- *
- * Test Configuration:
- * - high_threshold: 3 (trigger surge after 3 demand signals)
- * - surge_multiplier: 1.5x (50% price increase)
- * - low_threshold: 1 (trigger discount at 1 or fewer)
- * - discount_multiplier: 0.9x (10% discount)
- * - window_size: 10s (fast feedback loop)
- */
-
-import { test, expect, PricingAssertions } from '../lib/fixtures';
-import { EventNames, generateTestProductId } from '../lib/event-generator';
-
-test.describe('Dynamic Pricing Pipeline', () => {
-  test.describe.configure({ mode: 'serial' });
-
-  /**
-   * Scenario 1: Baseline Pricing
-   *
-   * Precondition: Clean state with no recent interactions for the product
-   * Expected: Price should equal base_price (markup = 1.0)
-   */
-  test('should return base price when no interactions exist', async ({ api, config }) => {
-    // Use a unique product ID to ensure no prior interactions
-    const productId = generateTestProductId('baseline');
-
-    // Get price from provider - should be base price (fallback)
-    // Note: This tests the fallback behavior when product isn't in Redis
-    const priceResponse = await api.getPrice('hotel', productId).catch(() => null);
-
-    // For unknown products, the API returns 404 or falls back to base
-    // This validates the fallback mechanism works
-    test.info().annotations.push({
-      type: 'info',
-      description: `Tested baseline pricing for product: ${productId}`,
-    });
-  });
-
-  /**
-   * Scenario 2: Surge Pricing Trigger
-   *
-   * Precondition: Fresh product with no interactions
-   * Action: Generate 5+ high-velocity interactions (above high_threshold=3)
-   * Expected: Price increases by surge_multiplier (1.5x)
-   */
-  test('should apply surge pricing when demand exceeds threshold', async ({
-    api,
-    events,
-    triggerPriceUpdate,
-    config,
-  }) => {
-    // Step 1: Create a fresh session
-    const sessionId = events.newSession();
-    const productId = generateTestProductId('surge');
-
-    test.info().annotations.push({
-      type: 'info',
-      description: `Testing surge pricing for product: ${productId}`,
-    });
-
-    // Step 2: Log initial price for this product (establish baseline)
-    await api.logPrice({
-      productId,
-      price: 100.0, // Base price
-      sessionId,
-      storeMode: 'hotel',
-    });
-
-    // Step 3: Generate high-velocity interactions (5 events > threshold of 3)
-    console.log(`\n📊 Generating ${5} surge events for product ${productId.slice(0, 8)}...`);
-
-    const surgeEvents = events.generateSurgeSequence(productId, 5);
-
-    for (const event of surgeEvents) {
-      await api.ingestEvent(event);
-      await new Promise(r => setTimeout(r, config.timing.eventDelay));
-    }
-
-    console.log(`✅ Ingested ${surgeEvents.length} events`);
-
-    // Step 4: Trigger the pricing pipeline
-    console.log('\n⚙️ Triggering pricing pipeline...');
-    const pipelineResult = await triggerPriceUpdate({
-      storeMode: 'hotel',
-      highThreshold: config.pricing.highThreshold,
-      surgeMultiplier: config.pricing.surgeMultiplier,
-    });
-
-    console.log(`📈 Pipeline processed ${pipelineResult.products_count} products`);
-
-    // Step 5: Verify surge pricing was applied
-    if (pipelineResult.prices && pipelineResult.prices.length > 0) {
-      const pricedProduct = pipelineResult.prices.find(p => p.productId === productId);
-
-      if (pricedProduct) {
-        const markup = pricedProduct.optimal_price / pricedProduct.base_price;
-
-        console.log(`\n💰 Price Result for ${productId.slice(0, 8)}:`);
-        console.log(`   Base Price: $${pricedProduct.base_price.toFixed(2)}`);
-        console.log(`   Optimal Price: $${pricedProduct.optimal_price.toFixed(2)}`);
-        console.log(`   Demand Score: ${pricedProduct.demand_score}`);
-        console.log(`   Markup: ${markup.toFixed(2)}x`);
-
-        // Verify surge was applied
-        expect(pricedProduct.demand_score).toBeGreaterThanOrEqual(config.pricing.highThreshold);
-        expect(markup).toBeCloseTo(config.pricing.surgeMultiplier, 1);
-      }
-    }
-
-    // Annotations for test report
-    test.info().annotations.push({
-      type: 'result',
-      description: `Pipeline processed ${pipelineResult.products_count} products`,
-    });
-  });
-
-  /**
-   * Scenario 3: Discount Pricing Trigger
-   *
-   * Precondition: Product with very low interaction count
-   * Action: Generate only 1 interaction (at or below low_threshold=1)
-   * Expected: Price decreases by discount_multiplier (0.9x)
-   */
-  test('should apply discount pricing when demand is below threshold', async ({
-    api,
-    events,
-    triggerPriceUpdate,
-    config,
-  }) => {
-    const sessionId = events.newSession();
-    const productId = generateTestProductId('discount');
-
-    test.info().annotations.push({
-      type: 'info',
-      description: `Testing discount pricing for product: ${productId}`,
-    });
-
-    // Step 1: Log initial price
-    await api.logPrice({
-      productId,
-      price: 100.0,
-      sessionId,
-      storeMode: 'hotel',
-    });
-
-    // Step 2: Generate minimal interaction (1 event = low_threshold)
-    console.log(`\n📊 Generating 1 low-demand event for product ${productId.slice(0, 8)}...`);
-
-    const event = events.viewProduct(productId);
-    await api.ingestEvent(event);
-
-    console.log('✅ Ingested 1 event');
-
-    // Step 3: Trigger pipeline
-    console.log('\n⚙️ Triggering pricing pipeline...');
-    const pipelineResult = await triggerPriceUpdate({
-      storeMode: 'hotel',
-      lowThreshold: config.pricing.lowThreshold,
-      discountMultiplier: config.pricing.discountMultiplier,
-    });
-
-    // Step 4: Verify discount pricing
-    if (pipelineResult.prices && pipelineResult.prices.length > 0) {
-      const pricedProduct = pipelineResult.prices.find(p => p.productId === productId);
-
-      if (pricedProduct) {
-        const markup = pricedProduct.optimal_price / pricedProduct.base_price;
-
-        console.log(`\n💰 Price Result for ${productId.slice(0, 8)}:`);
-        console.log(`   Base Price: $${pricedProduct.base_price.toFixed(2)}`);
-        console.log(`   Optimal Price: $${pricedProduct.optimal_price.toFixed(2)}`);
-        console.log(`   Demand Score: ${pricedProduct.demand_score}`);
-        console.log(`   Markup: ${markup.toFixed(2)}x`);
-
-        // Verify discount was applied
-        expect(pricedProduct.demand_score).toBeLessThanOrEqual(config.pricing.lowThreshold);
-        expect(markup).toBeCloseTo(config.pricing.discountMultiplier, 1);
-      }
-    }
-  });
-
-  /**
-   * Scenario 4: Multi-Product Differential Pricing
-   *
-   * Precondition: Multiple products with different interaction levels
-   * Action:
-   *   - Product A: 5 interactions (surge)
-   *   - Product B: 1 interaction (discount)
-   *   - Product C: 2 interactions (neutral)
-   * Expected: Each product priced according to its demand
-   */
-  test('should price multiple products differentially based on demand', async ({
-    api,
-    events,
-    triggerPriceUpdate,
-    config,
-  }) => {
-    const sessionId = events.newSession();
-
-    // Create 3 test products with different demand patterns
-    const products = {
-      surge: { id: generateTestProductId('multi-surge'), eventCount: 5, expectedMarkup: config.pricing.surgeMultiplier },
-      discount: { id: generateTestProductId('multi-discount'), eventCount: 1, expectedMarkup: config.pricing.discountMultiplier },
-      neutral: { id: generateTestProductId('multi-neutral'), eventCount: 2, expectedMarkup: 1.0 },
-    };
-
-    test.info().annotations.push({
-      type: 'info',
-      description: `Testing multi-product pricing: surge=${products.surge.id.slice(0, 8)}, discount=${products.discount.id.slice(0, 8)}, neutral=${products.neutral.id.slice(0, 8)}`,
-    });
-
-    // Step 1: Log base prices for all products
-    for (const [name, product] of Object.entries(products)) {
-      await api.logPrice({
-        productId: product.id,
-        price: 100.0,
-        sessionId,
-        storeMode: 'hotel',
-      });
-    }
-
-    // Step 2: Generate different interaction levels for each product
-    console.log('\n📊 Generating differentiated events:');
-
-    for (const [name, product] of Object.entries(products)) {
-      console.log(`   ${name}: ${product.eventCount} events`);
-
-      for (let i = 0; i < product.eventCount; i++) {
-        const event = events.viewProduct(product.id);
-        await api.ingestEvent(event);
-        await new Promise(r => setTimeout(r, 50));
-      }
-    }
-
-    console.log('✅ All events ingested');
-
-    // Step 3: Trigger pipeline
-    console.log('\n⚙️ Triggering pricing pipeline...');
-    const pipelineResult = await triggerPriceUpdate();
-
-    // Step 4: Verify differential pricing
-    console.log('\n💰 Multi-Product Pricing Results:');
-
-    if (pipelineResult.prices) {
-      for (const [name, product] of Object.entries(products)) {
-        const pricedProduct = pipelineResult.prices.find(p => p.productId === product.id);
-
-        if (pricedProduct) {
-          const markup = pricedProduct.optimal_price / pricedProduct.base_price;
-
-          console.log(`   ${name} (${product.id.slice(0, 8)}):`);
-          console.log(`      Demand: ${pricedProduct.demand_score}, Markup: ${markup.toFixed(2)}x (expected: ${product.expectedMarkup}x)`);
-
-          // Verify markup is in expected range (with tolerance)
-          expect(markup).toBeCloseTo(product.expectedMarkup, 1);
-        }
-      }
-    }
-  });
-
-  /**
-   * Scenario 5: Price Update Propagation
-   *
-   * Validates that price updates flow correctly from the pipeline
-   * through Redis to the Pricing Provider API.
-   */
-  test('should propagate prices from pipeline to pricing API', async ({
-    api,
-    events,
-    triggerPriceUpdate,
-    config,
-  }) => {
-    const sessionId = events.newSession();
-    const productId = generateTestProductId('propagation');
-
-    test.info().annotations.push({
-      type: 'info',
-      description: `Testing price propagation for product: ${productId}`,
-    });
-
-    // Step 1: Log base price
-    await api.logPrice({
-      productId,
-      price: 150.0, // Different base price for this test
-      sessionId,
-      storeMode: 'hotel',
-    });
-
-    // Step 2: Generate surge-level interactions
-    console.log(`\n📊 Generating surge events for propagation test...`);
-
-    const surgeEvents = events.generateSurgeSequence(productId, 6);
-    await api.ingestEvents(surgeEvents, config.timing.eventDelay);
-
-    console.log(`✅ Ingested ${surgeEvents.length} events`);
-
-    // Step 3: Trigger pipeline
-    console.log('\n⚙️ Triggering pricing pipeline...');
-    const pipelineResult = await triggerPriceUpdate();
-
-    expect(pipelineResult.success).toBe(true);
-    expect(pipelineResult.prices_published).toBe(true);
-
-    console.log(`📈 Pipeline published ${pipelineResult.products_count} prices to Redis`);
-
-    // Step 4: Wait for Redis propagation
-    await new Promise(r => setTimeout(r, 1000));
-
-    // Step 5: Verify via Pricing Provider API
-    // Note: This requires the product to exist in Supabase
-    // For pure E2E testing, we verify the pipeline output instead
-    if (pipelineResult.prices) {
-      const pricedProduct = pipelineResult.prices.find(p => p.productId === productId);
-
-      if (pricedProduct) {
-        console.log(`\n✅ Price Propagation Verified:`);
-        console.log(`   Product: ${productId.slice(0, 8)}`);
-        console.log(`   Base Price: $${pricedProduct.base_price.toFixed(2)}`);
-        console.log(`   Optimal Price: $${pricedProduct.optimal_price.toFixed(2)}`);
-        console.log(`   Published to Redis: ${pipelineResult.prices_published}`);
-
-        expect(pricedProduct.optimal_price).toBeGreaterThan(pricedProduct.base_price);
-      }
-    }
-  });
-
-  /**
-   * Scenario 6: Event Type Weighting
-   *
-   * Validates that different event types contribute to demand calculation.
-   * High-intent events (add_to_cart) should have more weight than low-intent (page_view).
-   */
-  test('should count various event types in demand calculation', async ({
-    api,
-    events,
-    triggerPriceUpdate,
-    config,
-  }) => {
-    const sessionId = events.newSession();
-    const productId = generateTestProductId('event-types');
-
-    test.info().annotations.push({
-      type: 'info',
-      description: `Testing event type weighting for product: ${productId}`,
-    });
-
-    // Log base price
-    await api.logPrice({
-      productId,
-      price: 100.0,
-      sessionId,
-      storeMode: 'hotel',
-    });
-
-    // Generate a mix of different event types
-    console.log('\n📊 Generating mixed event types:');
-
-    const mixedEvents = [
-      events.viewProduct(productId),           // page view
-      events.learnMore(productId),             // high intent
-      events.hover(productId, 'title'),        // engagement
-      events.hover(productId, 'paragraph'),    // engagement
-      events.addToCart(productId),             // highest intent
-    ];
-
-    console.log(`   - ${mixedEvents.length} mixed events (view, learn_more, hover, add_to_cart)`);
-
-    await api.ingestEvents(mixedEvents, config.timing.eventDelay);
-    console.log('✅ Events ingested');
-
-    // Trigger pipeline
-    const pipelineResult = await triggerPriceUpdate();
-
-    // Verify events were counted
-    if (pipelineResult.prices) {
-      const pricedProduct = pipelineResult.prices.find(p => p.productId === productId);
-
-      if (pricedProduct) {
-        console.log(`\n💰 Mixed Event Pricing Result:`);
-        console.log(`   Demand Score: ${pricedProduct.demand_score}`);
-        console.log(`   Expected: >= ${config.pricing.highThreshold} (for surge)`);
-
-        // Mixed events should trigger surge if count >= high_threshold
-        expect(pricedProduct.demand_score).toBeGreaterThanOrEqual(config.pricing.highThreshold);
-      }
-    }
-  });
-
-  /**
-   * Scenario 7: Session Isolation
-   *
-   * Validates that events from different sessions are correctly aggregated
-   * for the same product.
-   */
-  test('should aggregate demand across multiple sessions', async ({
-    api,
-    events,
-    triggerPriceUpdate,
-    config,
-  }) => {
-    const productId = generateTestProductId('multi-session');
-
-    test.info().annotations.push({
-      type: 'info',
-      description: `Testing multi-session aggregation for product: ${productId}`,
-    });
-
-    // Log base price
-    await api.logPrice({
-      productId,
-      price: 100.0,
-      sessionId: events.session,
-      storeMode: 'hotel',
-    });
-
-    // Generate events from 3 different sessions
-    console.log('\n📊 Generating events from multiple sessions:');
-
-    for (let i = 0; i < 3; i++) {
-      const sessionId = events.newSession();
-      console.log(`   Session ${i + 1}: ${sessionId.slice(0, 8)}...`);
-
-      // Each session generates 2 events
-      await api.ingestEvent(events.viewProduct(productId));
-      await api.ingestEvent(events.learnMore(productId));
-
-      await new Promise(r => setTimeout(r, config.timing.eventDelay));
-    }
-
-    console.log('✅ Events from 3 sessions ingested');
-
-    // Trigger pipeline
-    const pipelineResult = await triggerPriceUpdate();
-
-    // Verify aggregated demand
-    if (pipelineResult.prices) {
-      const pricedProduct = pipelineResult.prices.find(p => p.productId === productId);
-
-      if (pricedProduct) {
-        console.log(`\n💰 Multi-Session Aggregation Result:`);
-        console.log(`   Total Demand Score: ${pricedProduct.demand_score}`);
-        console.log(`   Expected: >= 6 (2 events × 3 sessions)`);
-
-        // 3 sessions × 2 events = 6 total events
-        expect(pricedProduct.demand_score).toBeGreaterThanOrEqual(6);
-      }
-    }
-  });
-});
-
-/**
- * Edge Cases and Error Handling
- */
-test.describe('Dynamic Pricing Edge Cases', () => {
-  test('should handle pipeline execution with empty Kafka topics', async ({
-    triggerPriceUpdate,
-  }) => {
-    // This tests the pipeline's resilience when there's no data
-    // The pipeline should complete without errors
-
-    console.log('\n⚙️ Testing pipeline with potentially empty data...');
-
-    // Run pipeline - should handle empty state gracefully
-    const result = await triggerPriceUpdate({ dryRun: true });
-
-    expect(result.success).toBe(true);
-    console.log(`✅ Pipeline handled gracefully: ${result.message || 'completed'}`);
-  });
-
-  test('should verify backend health before running tests', async ({ api }) => {
-    const backendHealth = await api.checkBackendHealth();
-    expect(backendHealth.status).toBe('healthy');
-
-    console.log(`✅ Backend: ${backendHealth.status}`);
-    console.log(`   Kafka: ${backendHealth.kafka}`);
-  });
-
-  test('should verify pricing provider health', async ({ api }) => {
-    const providerHealth = await api.checkProviderHealth();
-    expect(providerHealth.status).toBe('healthy');
-
-    console.log(`✅ Provider: ${providerHealth.status}`);
-    console.log(`   Redis: ${providerHealth.redis ? 'connected' : 'disconnected'}`);
-  });
-});

e2e/tsconfig.json

@@ -1,28 +0,0 @@
-{
-  "compilerOptions": {
-    "target": "ES2022",
-    "module": "NodeNext",
-    "moduleResolution": "NodeNext",
-    "strict": true,
-    "esModuleInterop": true,
-    "skipLibCheck": true,
-    "forceConsistentCasingInFileNames": true,
-    "resolveJsonModule": true,
-    "declaration": false,
-    "declarationMap": false,
-    "noEmit": true,
-    "outDir": "./dist",
-    "rootDir": ".",
-    "baseUrl": ".",
-    "paths": {
-      "@lib/*": ["lib/*"]
-    }
-  },
-  "include": [
-    "**/*.ts"
-  ],
-  "exclude": [
-    "node_modules",
-    "dist"
-  ]
-}

experiments/airflow/dags/elasticity_pricing_dag.py

@@ -0,0 +1,346 @@
+from airflow import DAG
+from airflow.operators.python import PythonOperator
+from airflow.utils.dates import days_ago
+from datetime import timedelta
+import pandas as pd
+import logging
+import sys
+import pickle
+import io
+
+# add parent dir to path so procesing package can be imported
+sys.path.insert(0, '/opt/airflow')
+
+from procesing.context import PipelineContext
+from procesing.providers import SupabaseProvider, BackendAPIProvider
+from procesing.steps import (
+    FetchInteractionsStep,
+    FetchPriceLogsStep,
+    CreatePriceBucketsStep,
+    AugmentEventNamesStep,
+    ChunkByTimeWindowStep,
+    ComputeDemandForChunksStep,
+    AggregatePriceLogsStep,
+    ComputeElasticityStep,
+    BuildStateSpaceStep,
+    FitPricingFunctionStep,
+    PredictPricesStep,
+)
+
+default_args = {
+    'owner': 'phantom-research',
+    'depends_on_past': False,
+    'email_on_failure': False,
+    'email_on_retry': False,
+    'retries': 2,
+    'retry_delay': timedelta(minutes=5),
+}
+
+def get_provider():
+    """Factory to create composite provider"""
+    class CompositeProvider(SupabaseProvider, BackendAPIProvider):
+        def __init__(self):
+            SupabaseProvider.__init__(self)
+            BackendAPIProvider.__init__(self)
+    return CompositeProvider()
+
+def get_context(**kwargs):
+    """Build pipeline context from Airflow config"""
+    dag_conf = kwargs.get('dag_run').conf if kwargs.get('dag_run') else {}
+    return PipelineContext(
+        provider=get_provider(),
+        store_mode=dag_conf.get('store_mode', 'hotel'),
+        window_size=dag_conf.get('window_size', '30s'),
+        n_price_buckets=dag_conf.get('n_price_buckets', 5),
+        elasticity_method=dag_conf.get('elasticity_method', 'point'),
+        min_observations=dag_conf.get('min_observations', 2),
+    )
+
+# atomic task functions (each wraps one sklearn step)
+def fetch_interactions(**kwargs):
+    """Task: Fetch interaction data from Kafka"""
+    context = get_context(**kwargs)
+    step = FetchInteractionsStep(context)
+    df = step.transform(None)
+
+    kwargs['ti'].xcom_push(key='interactions_raw', value=pickle.dumps(df))
+    logging.info(f"Fetched {len(df)} interaction records")
+    return len(df)
+
+def fetch_price_logs(**kwargs):
+    """Task: Fetch price logs from Kafka"""
+    context = get_context(**kwargs)
+    step = FetchPriceLogsStep(context)
+    df = step.transform(None)
+
+    kwargs['ti'].xcom_push(key='price_logs_raw', value=pickle.dumps(df))
+    logging.info(f"Fetched {len(df)} price records")
+    return len(df)
+
+def create_price_buckets(**kwargs):
+    """Task: Create price buckets for interactions"""
+    ti = kwargs['ti']
+    df = pickle.loads(ti.xcom_pull(key='interactions_raw'))
+
+    context = get_context(**kwargs)
+    step = CreatePriceBucketsStep(context)
+    df = step.transform(df)
+
+    ti.xcom_push(key='interactions_bucketed', value=pickle.dumps(df))
+    logging.info(f"Created price buckets for {len(df)} interactions")
+    return len(df)
+
+def augment_event_names(**kwargs):
+    """Task: Augment event names with product and price schema"""
+    ti = kwargs['ti']
+    df = pickle.loads(ti.xcom_pull(key='interactions_bucketed'))
+
+    context = get_context(**kwargs)
+    step = AugmentEventNamesStep(context)
+    df = step.transform(df)
+
+    ti.xcom_push(key='interactions_final', value=pickle.dumps(df))
+    logging.info(f"Augmented event names for {len(df)} interactions")
+    return len(df)
+
+def chunk_interactions(**kwargs):
+    """Task: Chunk interactions into time windows"""
+    ti = kwargs['ti']
+    df = pickle.loads(ti.xcom_pull(key='interactions_final'))
+
+    context = get_context(**kwargs)
+    step = ChunkByTimeWindowStep(context)
+    chunks = step.transform(df)
+
+    ti.xcom_push(key='interaction_chunks', value=pickle.dumps(chunks))
+    logging.info(f"Generated {len(chunks)} interaction chunks")
+    return len(chunks)
+
+def compute_demand(**kwargs):
+    """Task: Compute demand vectors for all chunks"""
+    ti = kwargs['ti']
+    chunks = pickle.loads(ti.xcom_pull(key='interaction_chunks'))
+
+    context = get_context(**kwargs)
+    step = ComputeDemandForChunksStep(context)
+    demand_chunks = step.transform(chunks)
+
+    ti.xcom_push(key='demand_chunks', value=pickle.dumps(demand_chunks))
+    logging.info(f"Computed demand for {len(demand_chunks)} chunks")
+    return len(demand_chunks)
+
+def aggregate_price_logs(**kwargs):
+    """Task: Aggregate price logs into time windows """
+    ti = kwargs['ti']
+    df = pickle.loads(ti.xcom_pull(key='price_logs_raw'))
+
+    context = get_context(**kwargs)
+    step = AggregatePriceLogsStep(context)
+    price_chunks = step.transform(df)
+
+    ti.xcom_push(key='price_chunks', value=pickle.dumps(price_chunks))
+    logging.info(f"Aggregated {len(price_chunks)} price chunks")
+    return len(price_chunks)
+
+def compute_elasticity(**kwargs):
+    """Task: Compute price elasticity from demand and price chunks"""
+    ti = kwargs['ti']
+    demand_chunks = pickle.loads(ti.xcom_pull(key='demand_chunks'))
+    price_chunks = pickle.loads(ti.xcom_pull(key='price_chunks'))
+
+    context = get_context(**kwargs)
+    step = ComputeElasticityStep(context)
+    elasticity_df = step.transform((demand_chunks, price_chunks))
+
+    ti.xcom_push(key='elasticity_results', value=pickle.dumps(elasticity_df))
+    logging.info(f"Computed elasticity for {len(elasticity_df)} products")
+
+    return {
+        'n_products': len(elasticity_df),
+        'mean_elasticity': float(elasticity_df['elasticity'].mean()),
+        'median_elasticity': float(elasticity_df['elasticity'].median())
+    }
+
+def build_state_space(**kwargs):
+    """Task: Build state space from elasticity"""
+    ti = kwargs['ti']
+    elasticity_df = pickle.loads(ti.xcom_pull(key='elasticity_results'))
+
+    context = get_context(**kwargs)
+    step = BuildStateSpaceStep(context)
+    state_space = step.transform(elasticity_df)
+
+    ti.xcom_push(key='state_space', value=pickle.dumps(state_space))
+    logging.info("Built state space for pricing")
+    return True
+
+def fit_pricing_function(**kwargs):
+    """Task: Fit pricing function using elasticity"""
+    ti = kwargs['ti']
+    elasticity_df = pickle.loads(ti.xcom_pull(key='elasticity_results'))
+
+    context = get_context(**kwargs)
+    step = FitPricingFunctionStep(context)
+    pricer = step.transform(elasticity_df)
+
+    ti.xcom_push(key='pricer', value=pickle.dumps(pricer))
+    logging.info("Fitted pricing function")
+    return True
+
+def predict_prices(**kwargs):
+    """Task: Predict optimal prices"""
+    ti = kwargs['ti']
+    pricer = pickle.loads(ti.xcom_pull(key='pricer'))
+    state_space = pickle.loads(ti.xcom_pull(key='state_space'))
+
+    context = get_context(**kwargs)
+    step = PredictPricesStep(context)
+    prices_df = step.transform((pricer, state_space))
+
+    ti.xcom_push(key='predicted_prices', value=pickle.dumps(prices_df))
+    logging.info(f"Predicted prices for {len(prices_df)} products")
+    return len(prices_df)
+
+def publish_results(**kwargs):
+    """Task: Publish elasticity and pricing results to model registry"""
+    ti = kwargs['ti']
+    elasticity_df = pickle.loads(ti.xcom_pull(key='elasticity_results'))
+    prices_df = pickle.loads(ti.xcom_pull(key='predicted_prices'))
+
+    sys.path.insert(0, '/opt/airflow')
+    from lib.model_registry import ModelRegistry
+
+    registry = ModelRegistry()
+    dag_conf = kwargs.get('dag_run').conf if kwargs.get('dag_run') else {}
+
+    metadata = {
+        'timestamp': pd.Timestamp.now().isoformat(),
+        'window_size': dag_conf.get('window_size', '30s'),
+        'store_mode': dag_conf.get('store_mode', 'hotel'),
+        'dag_run_id': kwargs['dag_run'].run_id if kwargs.get('dag_run') else 'manual'
+    }
+
+    registry.publish_elasticity(elasticity_df, model_name='latest', metadata=metadata)
+
+    # get fitted pricer from XCom
+    pricer = pickle.loads(ti.xcom_pull(key='pricer'))
+    registry.publish_pricing_model(
+        pricer,
+        model_name='latest',
+        metadata={**metadata, 'model_type': type(pricer).__name__}
+    )
+
+    logging.info(f"Published elasticity + pricing for {len(elasticity_df)} products")
+
+    return {
+        'n_products': len(elasticity_df),
+        'registry_status': 'success',
+        'elasticity_mean': float(elasticity_df['elasticity'].mean())
+    }
+
+
+# DAG definition
+with DAG(
+    'elasticity_pricing_pipeline',
+    default_args=default_args,
+    description='E2E refactored pipeline: atomic steps with proper separation',
+    schedule_interval='*/15 * * * *',
+    start_date=days_ago(1),
+    catchup=False,
+    max_active_runs=1,
+    tags=['pricing', 'elasticity', 'research', 'refactored'],
+) as dag:
+
+    # parallel data fetching
+    t_fetch_interactions = PythonOperator(
+        task_id='fetch_interactions',
+        python_callable=fetch_interactions,
+        provide_context=True,
+    )
+
+    t_fetch_price_logs = PythonOperator(
+        task_id='fetch_price_logs',
+        python_callable=fetch_price_logs,
+        provide_context=True,
+    )
+
+    # interaction processing branch
+    t_create_buckets = PythonOperator(
+        task_id='create_price_buckets',
+        python_callable=create_price_buckets,
+        provide_context=True,
+    )
+
+    t_augment_events = PythonOperator(
+        task_id='augment_event_names',
+        python_callable=augment_event_names,
+        provide_context=True,
+    )
+
+    t_chunk_interactions = PythonOperator(
+        task_id='chunk_interactions',
+        python_callable=chunk_interactions,
+        provide_context=True,
+    )
+
+    t_compute_demand = PythonOperator(
+        task_id='compute_demand',
+        python_callable=compute_demand,
+        provide_context=True,
+    )
+
+    # price processing branch (VECTORIZED)
+    t_aggregate_prices = PythonOperator(
+        task_id='aggregate_price_logs',
+        python_callable=aggregate_price_logs,
+        provide_context=True,
+    )
+
+    # convergence: compute elasticity
+    t_compute_elasticity = PythonOperator(
+        task_id='compute_elasticity',
+        python_callable=compute_elasticity,
+        provide_context=True,
+    )
+
+    # pricing tasks
+    t_build_state = PythonOperator(
+        task_id='build_state_space',
+        python_callable=build_state_space,
+        provide_context=True,
+    )
+
+    t_fit_pricer = PythonOperator(
+        task_id='fit_pricing_function',
+        python_callable=fit_pricing_function,
+        provide_context=True,
+    )
+
+    t_predict_prices = PythonOperator(
+        task_id='predict_prices',
+        python_callable=predict_prices,
+        provide_context=True,
+    )
+
+    # publish to registry
+    t_publish = PythonOperator(
+        task_id='publish_results',
+        python_callable=publish_results,
+        provide_context=True,
+    )
+
+    # dependency graph (clear atomic flow)
+    # parallel fetches
+    [t_fetch_interactions, t_fetch_price_logs]
+
+    # interaction branch: fetch -> bucket -> augment -> chunk -> demand
+    t_fetch_interactions >> t_create_buckets >> t_augment_events >> t_chunk_interactions >> t_compute_demand
+
+    # price branch: fetch -> aggregate (vectorized)
+    t_fetch_price_logs >> t_aggregate_prices
+
+    # convergence: both branches -> elasticity
+    [t_compute_demand, t_aggregate_prices] >> t_compute_elasticity
+
+    # pricing: elasticity -> state + fit -> predict -> publish
+    t_compute_elasticity >> [t_build_state, t_fit_pricer] >> t_predict_prices >> t_publish

experiments/airflow/dags/ml_training_pipeline.py

@@ -1,115 +0,0 @@
-from airflow import DAG, Dataset
-from airflow.decorators import task
-from airflow.utils.dates import days_ago
-from datetime import timedelta
-import pandas as pd
-import logging
-import sys
-import pickle
-
-sys.path.insert(0, '/opt/airflow')
-
-from procesing.context import PipelineContext
-from procesing.providers import SupabaseProvider, BackendAPIProvider
-from procesing.steps import (
-    FetchInteractionsStep,
-    ValidateDataStep,
-    ExtractSessionFeaturesStep,
-    JoinLabelsStep,
-)
-
-TRAINING_DATASET = Dataset('phantom://ml/training-data')
-
-DEFAULT_ARGS = {
-    'owner': 'phantom-research',
-    'depends_on_past': False,
-    'email_on_failure': False,
-    'email_on_retry': False,
-    'retries': 2,
-    'retry_delay': timedelta(minutes=5),
-}
-
-
-class CompositeProvider(SupabaseProvider, BackendAPIProvider):
-    def __init__(self):
-        SupabaseProvider.__init__(self)
-        BackendAPIProvider.__init__(self)
-
-
-def _get_context(store_mode: str = 'hotel') -> PipelineContext:
-    return PipelineContext(provider=CompositeProvider(), store_mode=store_mode)
-
-
-with DAG(
-    'ml_training_pipeline',
-    default_args=DEFAULT_ARGS,
-    description='ML training data pipeline: fetch -> validate -> extract features -> label -> publish',
-    schedule=None,
-    start_date=days_ago(1),
-    catchup=False,
-    max_active_runs=1,
-    tags=['ml', 'training', 'features', 'research'],
-) as dag:
-
-    @task
-    def fetch_interactions(**kwargs) -> bytes:
-        dag_conf = kwargs.get('dag_run').conf if kwargs.get('dag_run') else {}
-        ctx = _get_context(dag_conf.get('store_mode', 'hotel'))
-        df = FetchInteractionsStep(ctx).transform(None)
-        logging.info(f"Fetched {len(df)} interactions, {df['sessionId'].nunique()} sessions")
-        return pickle.dumps(df)
-
-    @task
-    def validate_data(raw_data: bytes, **kwargs) -> bytes:
-        df = pickle.loads(raw_data)
-        dag_conf = kwargs.get('dag_run').conf if kwargs.get('dag_run') else {}
-        ctx = _get_context(dag_conf.get('store_mode', 'hotel'))
-        validated = ValidateDataStep(ctx).transform(df)
-        report = ctx.get_cached('validation_report') or {}
-        logging.info(f"Validation: {report.get('status')}, {report.get('sessions', 0)} sessions")
-        return pickle.dumps(validated)
-
-    @task
-    def extract_session_features(validated_data: bytes, **kwargs) -> bytes:
-        df = pickle.loads(validated_data)
-        if df.empty:
-            logging.warning("Empty input, skipping feature extraction")
-            return pickle.dumps(pd.DataFrame())
-        dag_conf = kwargs.get('dag_run').conf if kwargs.get('dag_run') else {}
-        ctx = _get_context(dag_conf.get('store_mode', 'hotel'))
-        features = ExtractSessionFeaturesStep(ctx).transform(df)
-        logging.info(f"Extracted {len(features.columns)} features for {len(features)} sessions")
-        return pickle.dumps(features)
-
-    @task
-    def join_labels(features_data: bytes, **kwargs) -> bytes:
-        features_df = pickle.loads(features_data)
-        if features_df.empty:
-            logging.warning("Empty features, skipping label join")
-            return pickle.dumps(pd.DataFrame())
-        dag_conf = kwargs.get('dag_run').conf if kwargs.get('dag_run') else {}
-        ctx = _get_context(dag_conf.get('store_mode', 'hotel'))
-        labeled = JoinLabelsStep(ctx).transform(features_df)
-        n_agents = labeled['is_agent'].sum() if 'is_agent' in labeled.columns else 0
-        logging.info(f"Labeled {len(labeled)} sessions: {n_agents} agents")
-        return pickle.dumps(labeled)
-
-    @task(outlets=[TRAINING_DATASET])
-    def publish_training_data(labeled_data: bytes, **kwargs) -> dict:
-        labeled_df = pickle.loads(labeled_data)
-        if labeled_df.empty:
-            return {'status': 'skipped', 'reason': 'empty_data'}
-        dag_conf = kwargs.get('dag_run').conf if kwargs.get('dag_run') else {}
-        return {
-            'status': 'success',
-            'n_sessions': len(labeled_df),
-            'n_features': len([c for c in labeled_df.columns if c not in ['sessionId', 'experimentId', 'is_agent']]),
-            'store_mode': dag_conf.get('store_mode', 'hotel'),
-            'timestamp': pd.Timestamp.now().isoformat(),
-        }
-
-    raw = fetch_interactions()
-    validated = validate_data(raw)
-    features = extract_session_features(validated)
-    labeled = join_labels(features)
-    publish_training_data(labeled)

experiments/airflow/dags/surge_pricing_factory.py

@@ -1,210 +0,0 @@
-from airflow import DAG
-from airflow.operators.python import PythonOperator
-from airflow.utils.dates import days_ago
-from datetime import timedelta
-import pandas as pd
-import logging
-import sys
-import pickle
-
-sys.path.insert(0, '/opt/airflow')
-
-from procesing.context import PipelineContext
-from procesing.providers import SupabaseProvider, BackendAPIProvider
-from procesing.steps import (
-    FetchInteractionsStep,
-    FetchPriceLogsStep,
-    ComputeDemandStep,
-    AggregatePriceLogsStep,
-    JoinProductFeaturesStep,
-)
-from procesing.pricers.simple import SimpleSurgePricer
-
-DEFAULT_ARGS = {
-    'owner': 'phantom-research',
-    'depends_on_past': False,
-    'email_on_failure': False,
-    'email_on_retry': False,
-    'retries': 2,
-    'retry_delay': timedelta(minutes=5),
-}
-
-class CompositeProvider(SupabaseProvider, BackendAPIProvider):
-    def __init__(self):
-        SupabaseProvider.__init__(self)
-        BackendAPIProvider.__init__(self)
-
-def _get_provider():
-    return CompositeProvider()
-
-def _make_task_callables(store_mode: str):
-    """Generate task callables bound to a specific store_mode."""
-
-    def get_context(**kwargs):
-        return PipelineContext(provider=_get_provider(), store_mode=store_mode)
-
-    def fetch_interactions(**kwargs):
-        ctx = get_context(**kwargs)
-        df = FetchInteractionsStep(ctx).transform(None)
-        kwargs['ti'].xcom_push(key='interactions_raw', value=pickle.dumps(df))
-        logging.info(f"[{store_mode}] Fetched {len(df)} interaction records")
-        return len(df)
-
-    def fetch_price_logs(**kwargs):
-        ctx = get_context(**kwargs)
-        df = FetchPriceLogsStep(ctx).transform(None)
-        kwargs['ti'].xcom_push(key='price_logs_raw', value=pickle.dumps(df))
-        logging.info(f"[{store_mode}] Fetched {len(df)} price records")
-        return len(df)
-
-    def compute_demand(**kwargs):
-        ti = kwargs['ti']
-        df = pickle.loads(ti.xcom_pull(key='interactions_raw'))
-        ctx = get_context(**kwargs)
-        demand_df = ComputeDemandStep(ctx).transform(df)
-        ti.xcom_push(key='demand_data', value=pickle.dumps(demand_df))
-        logging.info(f"[{store_mode}] Computed demand for {len(demand_df)} products")
-        return len(demand_df)
-
-    def aggregate_price_logs(**kwargs):
-        ti = kwargs['ti']
-        df = pickle.loads(ti.xcom_pull(key='price_logs_raw'))
-        ctx = get_context(**kwargs)
-        price_df = AggregatePriceLogsStep(ctx).transform(df)
-        ti.xcom_push(key='price_data', value=pickle.dumps(price_df))
-        logging.info(f"[{store_mode}] Aggregated price logs for {len(price_df)} products")
-        return len(price_df)
-
-    def join_product_features(**kwargs):
-        ti = kwargs['ti']
-        demand_df = pickle.loads(ti.xcom_pull(key='demand_data'))
-        price_df = pickle.loads(ti.xcom_pull(key='price_data'))
-        ctx = get_context(**kwargs)
-        joined_df = JoinProductFeaturesStep(ctx).transform((demand_df, price_df))
-        ti.xcom_push(key='product_features', value=pickle.dumps(joined_df))
-        logging.info(f"[{store_mode}] Joined features for {len(joined_df)} products")
-        return len(joined_df)
-
-    def apply_surge_pricing(**kwargs):
-        ti = kwargs['ti']
-        product_features = pickle.loads(ti.xcom_pull(key='product_features'))
-        dag_conf = kwargs.get('dag_run').conf if kwargs.get('dag_run') else {}
-
-        data = product_features.rename(columns={'demand_score': 'demand'})
-        surge_pricer = SimpleSurgePricer(
-            high_threshold=dag_conf.get('high_threshold', 10),
-            low_threshold=dag_conf.get('low_threshold', 2),
-            surge_multiplier=dag_conf.get('surge_multiplier', 1.2),
-            discount_multiplier=dag_conf.get('discount_multiplier', 0.9)
-        )
-        surge_pricer.fit(data)
-        data['optimal_price'] = surge_pricer.predict()
-
-        prices_df = data[['productId', 'price', 'base_price', 'optimal_price', 'demand']].rename(columns={
-            'price': 'current_price', 'demand': 'demand_score'
-        })
-        ti.xcom_push(key='predicted_prices', value=pickle.dumps(prices_df))
-        logging.info(f"[{store_mode}] Applied surge pricing for {len(prices_df)} products")
-        return len(prices_df)
-
-    def publish_results(**kwargs):
-        ti = kwargs['ti']
-        prices_df = pickle.loads(ti.xcom_pull(key='predicted_prices'))
-        from lib.model_registry import ModelRegistry
-
-        registry = ModelRegistry()
-        dag_conf = kwargs.get('dag_run').conf if kwargs.get('dag_run') else {}
-
-        metadata = {
-            'timestamp': pd.Timestamp.now().isoformat(),
-            'store_mode': store_mode,
-            'dag_run_id': kwargs['dag_run'].run_id if kwargs.get('dag_run') else 'manual',
-            'pricing_method': 'surge',
-            'high_threshold': dag_conf.get('high_threshold', 10),
-            'low_threshold': dag_conf.get('low_threshold', 2),
-            'surge_multiplier': dag_conf.get('surge_multiplier', 1.2),
-            'discount_multiplier': dag_conf.get('discount_multiplier', 0.9)
-        }
-        registry.publish_prices(prices_df, model_name=f'{store_mode}_latest', metadata=metadata)
-        logging.info(f"[{store_mode}] Published surge pricing for {len(prices_df)} products")
-
-        return {
-            'n_products': len(prices_df),
-            'registry_status': 'success',
-            'store_mode': store_mode,
-            'mean_demand': float(prices_df['demand_score'].mean()) if 'demand_score' in prices_df.columns else None
-        }
-
-    return {
-        'fetch_interactions': fetch_interactions,
-        'fetch_price_logs': fetch_price_logs,
-        'compute_demand': compute_demand,
-        'aggregate_price_logs': aggregate_price_logs,
-        'join_product_features': join_product_features,
-        'apply_surge_pricing': apply_surge_pricing,
-        'publish_results': publish_results,
-    }
-
-
-def create_surge_pricing_dag(store_mode: str) -> DAG:
-    """Factory: generates a surge pricing DAG for a given store_mode."""
-    callables = _make_task_callables(store_mode)
-
-    dag = DAG(
-        f'surge_pricing_{store_mode}',
-        default_args=DEFAULT_ARGS,
-        description=f'Surge pricing pipeline for {store_mode} store mode',
-        schedule_interval='*/15 * * * *',
-        start_date=days_ago(1),
-        catchup=False,
-        max_active_runs=1,
-        tags=['pricing', 'surge', 'research', store_mode],
-    )
-
-    with dag:
-        t_fetch_interactions = PythonOperator(
-            task_id='fetch_interactions',
-            python_callable=callables['fetch_interactions'],
-            provide_context=True,
-        )
-        t_fetch_price_logs = PythonOperator(
-            task_id='fetch_price_logs',
-            python_callable=callables['fetch_price_logs'],
-            provide_context=True,
-        )
-        t_compute_demand = PythonOperator(
-            task_id='compute_demand',
-            python_callable=callables['compute_demand'],
-            provide_context=True,
-        )
-        t_aggregate_prices = PythonOperator(
-            task_id='aggregate_price_logs',
-            python_callable=callables['aggregate_price_logs'],
-            provide_context=True,
-        )
-        t_join_features = PythonOperator(
-            task_id='join_product_features',
-            python_callable=callables['join_product_features'],
-            provide_context=True,
-        )
-        t_surge_pricing = PythonOperator(
-            task_id='apply_surge_pricing',
-            python_callable=callables['apply_surge_pricing'],
-            provide_context=True,
-        )
-        t_publish = PythonOperator(
-            task_id='publish_results',
-            python_callable=callables['publish_results'],
-            provide_context=True,
-        )
-
-        t_fetch_interactions >> t_compute_demand
-        t_fetch_price_logs >> t_aggregate_prices
-        [t_compute_demand, t_aggregate_prices] >> t_join_features >> t_surge_pricing >> t_publish
-
-    return dag
-
-
-# instantiate DAGs for Airflow to discover
-dag_airline = create_surge_pricing_dag('airline')
-dag_hotel = create_surge_pricing_dag('hotel')

experiments/airflow/dags/surge_pricing_pipeline.py

@@ -1,237 +0,0 @@
-from airflow import DAG
-from airflow.operators.python import PythonOperator
-from airflow.utils.dates import days_ago
-from datetime import timedelta
-import pandas as pd
-import logging
-import sys
-import pickle
-import io
-
-# add parent dir to path so procesing package can be imported
-sys.path.insert(0, '/opt/airflow')
-
-from procesing.context import PipelineContext
-from procesing.providers import SupabaseProvider, BackendAPIProvider
-from procesing.steps import (
-    FetchInteractionsStep,
-    FetchPriceLogsStep,
-    ComputeDemandStep,
-    AggregatePriceLogsStep,
-    JoinProductFeaturesStep,
-)
-from procesing.pricers.simple import SimpleSurgePricer
-
-default_args = {
-    'owner': 'phantom-research',
-    'depends_on_past': False,
-    'email_on_failure': False,
-    'email_on_retry': False,
-    'retries': 2,
-    'retry_delay': timedelta(minutes=5),
-}
-
-def get_provider():
-    """Factory to create composite provider"""
-    class CompositeProvider(SupabaseProvider, BackendAPIProvider): # TODO: Fix this into one global provider singelton instead of multiple inheritance declarations acoss the codebase
-        def __init__(self):
-            SupabaseProvider.__init__(self)
-            BackendAPIProvider.__init__(self)
-    return CompositeProvider()
-
-def get_context(**kwargs):
-    """Build pipeline context from Airflow config"""
-    dag_conf = kwargs.get('dag_run').conf if kwargs.get('dag_run') else {}
-    return PipelineContext(
-        provider=get_provider(),
-        store_mode=dag_conf.get('store_mode', 'hotel'),
-    )
-
-# atomic task functions (each wraps one sklearn step)
-def fetch_interactions(**kwargs):
-    """Task: Fetch interaction data from Kafka"""
-    context = get_context(**kwargs)
-    step = FetchInteractionsStep(context)
-    df = step.transform(None)
-
-    kwargs['ti'].xcom_push(key='interactions_raw', value=pickle.dumps(df))
-    logging.info(f"Fetched {len(df)} interaction records")
-    return len(df)
-
-def fetch_price_logs(**kwargs):
-    """Task: Fetch price logs from Kafka"""
-    context = get_context(**kwargs)
-    step = FetchPriceLogsStep(context)
-    df = step.transform(None)
-
-    kwargs['ti'].xcom_push(key='price_logs_raw', value=pickle.dumps(df))
-    logging.info(f"Fetched {len(df)} price records")
-    return len(df)
-
-def compute_demand(**kwargs):
-    """Task: Compute demand scores from interactions"""
-    ti = kwargs['ti']
-    df = pickle.loads(ti.xcom_pull(key='interactions_raw'))
-
-    context = get_context(**kwargs)
-    step = ComputeDemandStep(context)
-    demand_df = step.transform(df)
-    # TODO: clear the xcom
-
-
-    ti.xcom_push(key='demand_data', value=pickle.dumps(demand_df))
-    logging.info(f"Computed demand for {len(demand_df)} products")
-    return len(demand_df)
-
-def aggregate_price_logs(**kwargs):
-    """Task: Aggregate price logs"""
-    ti = kwargs['ti']
-    df = pickle.loads(ti.xcom_pull(key='price_logs_raw'))
-
-    context = get_context(**kwargs)
-    step = AggregatePriceLogsStep(context)
-    price_df = step.transform(df)
-
-    ti.xcom_push(key='price_data', value=pickle.dumps(price_df))
-    logging.info(f"Aggregated price logs for {len(price_df)} products")
-    return len(price_df)
-
-def join_product_features(**kwargs):
-    """Task: Join demand and price data"""
-    ti = kwargs['ti']
-    demand_df = pickle.loads(ti.xcom_pull(key='demand_data'))
-    price_df = pickle.loads(ti.xcom_pull(key='price_data'))
-
-    context = get_context(**kwargs)
-    step = JoinProductFeaturesStep(context)
-    joined_df = step.transform((demand_df, price_df))
-
-    ti.xcom_push(key='product_features', value=pickle.dumps(joined_df))
-    logging.info(f"Joined features for {len(joined_df)} products")
-    return len(joined_df)
-
-def apply_surge_pricing(**kwargs):
-    """Task: Apply surge pricing rules to generate optimal prices"""
-    ti = kwargs['ti']
-    product_features = pickle.loads(ti.xcom_pull(key='product_features'))
-
-    dag_conf = kwargs.get('dag_run').conf if kwargs.get('dag_run') else {}
-
-    # rename demand_score to demand for pricer compatibility
-    data = product_features.rename(columns={'demand_score': 'demand'})
-
-    surge_pricer = SimpleSurgePricer(
-        high_threshold=dag_conf.get('high_threshold', 10),
-        low_threshold=dag_conf.get('low_threshold', 2),
-        surge_multiplier=dag_conf.get('surge_multiplier', 1.2),
-        discount_multiplier=dag_conf.get('discount_multiplier', 0.9)
-    )
-    surge_pricer.fit(data)
-    data['optimal_price'] = surge_pricer.predict()
-
-    prices_df = data[['productId', 'price', 'base_price', 'optimal_price', 'demand']].rename(columns={
-        'price': 'current_price',
-        'demand': 'demand_score'
-    })
-
-    ti.xcom_push(key='predicted_prices', value=pickle.dumps(prices_df))
-    logging.info(f"Applied surge pricing for {len(prices_df)} products")
-    return len(prices_df)
-
-def publish_results(**kwargs):
-    """Task: Publish surge pricing results to registry"""
-    ti = kwargs['ti']
-    prices_df = pickle.loads(ti.xcom_pull(key='predicted_prices'))
-
-    sys.path.insert(0, '/opt/airflow')
-    from lib.model_registry import ModelRegistry
-
-    registry = ModelRegistry()
-    dag_conf = kwargs.get('dag_run').conf if kwargs.get('dag_run') else {}
-
-    metadata = {
-        'timestamp': pd.Timestamp.now().isoformat(),
-        'store_mode': dag_conf.get('store_mode', 'hotel'),
-        'dag_run_id': kwargs['dag_run'].run_id if kwargs.get('dag_run') else 'manual',
-        'pricing_method': 'surge',
-        'high_threshold': dag_conf.get('high_threshold', 10),
-        'low_threshold': dag_conf.get('low_threshold', 2),
-        'surge_multiplier': dag_conf.get('surge_multiplier', 1.2),
-        'discount_multiplier': dag_conf.get('discount_multiplier', 0.9)
-    }
-
-    registry.publish_prices(prices_df, model_name='latest', metadata=metadata)
-
-    logging.info(f"Published surge pricing for {len(prices_df)} products")
-
-    return {
-        'n_products': len(prices_df),
-        'registry_status': 'success',
-        'mean_demand': float(prices_df['demand_score'].mean()) if 'demand_score' in prices_df.columns else None
-    }
-
-
-# DAG definition
-with DAG(
-    'surge_pricing_pipeline',
-    default_args=default_args,
-    description='Simple surge pricing pipeline: demand aggregation + rule-based pricing',
-    schedule_interval='*/15 * * * *',
-    start_date=days_ago(1),
-    catchup=False,
-    max_active_runs=1,
-    tags=['pricing', 'surge', 'research', 'simplified'],
-) as dag:
-
-    # parallel data fetching
-    t_fetch_interactions = PythonOperator(
-        task_id='fetch_interactions',
-        python_callable=fetch_interactions,
-        provide_context=True,
-    )
-
-    t_fetch_price_logs = PythonOperator(
-        task_id='fetch_price_logs',
-        python_callable=fetch_price_logs,
-        provide_context=True,
-    )
-
-    # compute demand from interactions
-    t_compute_demand = PythonOperator(
-        task_id='compute_demand',
-        python_callable=compute_demand,
-        provide_context=True,
-    )
-
-    # aggregate price logs
-    t_aggregate_prices = PythonOperator(
-        task_id='aggregate_price_logs',
-        python_callable=aggregate_price_logs,
-        provide_context=True,
-    )
-
-    # join demand and prices
-    t_join_features = PythonOperator(
-        task_id='join_product_features',
-        python_callable=join_product_features,
-        provide_context=True,
-    )
-
-    # apply surge pricing
-    t_surge_pricing = PythonOperator(
-        task_id='apply_surge_pricing',
-        python_callable=apply_surge_pricing,
-        provide_context=True,
-    )
-
-    # publish to registry
-    t_publish = PythonOperator(
-        task_id='publish_results',
-        python_callable=publish_results,
-        provide_context=True,
-    )
-
-    # dependency graph: parallel fetch -> process -> join -> surge -> publish
-    t_fetch_interactions >> t_compute_demand
-    t_fetch_price_logs >> t_aggregate_prices
-    [t_compute_demand, t_aggregate_prices] >> t_join_features >> t_surge_pricing >> t_publish

experiments/ml/__init__.py

@@ -1,11 +0,0 @@
-from .evals import evaluate
-from .arch import (
-    XGBoostAgentClassifier,
-    LightGBMAgentClassifier
-)
-
-__all__ =[
-    'evaluate',
-    'XGBoostAgentClassifier',
-    'LightGBMAgentClassifier'
-]

experiments/ml/arch.py

@@ -1,122 +0,0 @@
-# sklearn compatible models for agent detection
-from sklearn.base import BaseEstimator, ClassifierMixin
-from procesing.context import PipelineContext
-from typing import Any, Optional, Tuple
-from abc import ABC, abstractmethod
-import xgboost as xgb
-import lightgbm as lgb
-import numpy as np
-import pandas as pd
-
-TASK = 'classification'
-LABELS = ['human', 'agent']
-
-
-class BaseAgentClassifier(BaseEstimator, ClassifierMixin, ABC):
-    """Base class for tree-based agent detection classifiers with common logic"""
-
-    def __init__(self, context: Optional[PipelineContext] = None, n_estimators: int = 200,
-                 max_depth: int = 6, learning_rate: float = 0.05,
-                 early_stopping_rounds: int = 20):
-        self.context = context
-        self.n_estimators = n_estimators
-        self.max_depth = max_depth
-        self.learning_rate = learning_rate
-        self.early_stopping_rounds = early_stopping_rounds
-        self.model_ = None
-        self.feature_names_ = None
-
-    def _to_array(self, X):
-        """Convert pandas structures to numpy arrays"""
-        return X.values if isinstance(X, (pd.DataFrame, pd.Series)) else X
-
-    def _compute_pos_weight(self, y_arr):
-        """Calculate scale_pos_weight for class imbalance handling"""
-        n_neg, n_pos = (y_arr == 0).sum(), (y_arr == 1).sum()
-        return n_neg / n_pos if n_pos > 0 else 1.0
-
-    def _prepare_eval_set(self, eval_set):
-        """Convert eval_set to numpy arrays if needed"""
-        if not eval_set:
-            return None
-        X_val, y_val = eval_set[0]
-        return [(self._to_array(X_val), self._to_array(y_val))]
-
-    @abstractmethod
-    def _build_model(self, scale_pos: float):
-        """Build the underlying model instance (must be implemented by subclasses)"""
-        pass
-
-    @abstractmethod
-    def _fit_with_eval(self, X_arr, y_arr, eval_arr):
-        """Fit model with evaluation set (must be implemented by subclasses)"""
-        pass
-
-    def fit(self, X, y, eval_set=None):
-        X_arr, y_arr = self._to_array(X), self._to_array(y)
-
-        if isinstance(X, pd.DataFrame):
-            self.feature_names_ = X.columns.tolist()
-
-        scale_pos = self._compute_pos_weight(y_arr)
-        self.model_ = self._build_model(scale_pos)
-
-        eval_arr = self._prepare_eval_set(eval_set)
-        if eval_arr:
-            self._fit_with_eval(X_arr, y_arr, eval_arr)
-        else:
-            self.model_.fit(X_arr, y_arr)
-
-        return self
-
-    def predict(self, X):
-        return self.model_.predict(self._to_array(X))
-
-    def predict_proba(self, X):
-        return self.model_.predict_proba(self._to_array(X))
-
-    @property
-    def feature_importances_(self):
-        return self.model_.feature_importances_ if self.model_ else None
-
-
-class XGBoostAgentClassifier(BaseAgentClassifier):
-    """XGBoost binary classifier for agent detection with class imbalance handling"""
-
-    def _build_model(self, scale_pos: float):
-        return xgb.XGBClassifier(
-            n_estimators=self.n_estimators,
-            max_depth=self.max_depth,
-            learning_rate=self.learning_rate,
-            scale_pos_weight=scale_pos,
-            eval_metric='auc',
-            early_stopping_rounds=self.early_stopping_rounds,
-            random_state=42,
-            tree_method='hist',
-            enable_categorical=False
-        )
-
-    def _fit_with_eval(self, X_arr, y_arr, eval_arr):
-        self.model_.fit(X_arr, y_arr, eval_set=eval_arr, verbose=False)
-
-
-class LightGBMAgentClassifier(BaseAgentClassifier):
-    """LightGBM binary classifier for agent detection with class imbalance handling"""
-
-    def _build_model(self, scale_pos: float):
-        return lgb.LGBMClassifier(
-            n_estimators=self.n_estimators,
-            max_depth=self.max_depth,
-            learning_rate=self.learning_rate,
-            scale_pos_weight=scale_pos,
-            metric='auc',
-            random_state=42,
-            verbosity=-1
-        )
-
-    def _fit_with_eval(self, X_arr, y_arr, eval_arr):
-        self.model_.fit(
-            X_arr, y_arr,
-            eval_set=eval_arr,
-            callbacks=[lgb.early_stopping(self.early_stopping_rounds, verbose=False)]
-        )

experiments/ml/evals.py

@@ -1,103 +0,0 @@
-from sklearn.metrics import (accuracy_score, precision_score, recall_score,
-                             f1_score, roc_auc_score, confusion_matrix, roc_curve)
-from torch.utils.tensorboard import SummaryWriter
-from logging import getLogger
-import numpy as np
-import matplotlib.pyplot as plt
-import io
-from PIL import Image
-
-logger = getLogger(__name__)
-
-
-def log_feature_importance(writer, model, feature_names, epoch):
-    """Visualize and log feature importance to TensorBoard"""
-    if not hasattr(model, 'feature_importances_') or model.feature_importances_ is None:
-        return
-
-    importance = model.feature_importances_
-    indices = np.argsort(importance)[::-1][:20]  # top 20
-    top_features = [feature_names[i] for i in indices]
-    top_importance = importance[indices]
-
-    for i, (feat, imp) in enumerate(zip(top_features, top_importance)):
-        writer.add_scalar(f'FeatureImportance/{feat}', imp, epoch)
-
-    fig, ax = plt.subplots(figsize=(10, 8))
-    ax.barh(range(len(top_features)), top_importance, align='center')
-    ax.set_yticks(range(len(top_features)))
-    ax.set_yticklabels(top_features)
-    ax.invert_yaxis()
-    ax.set_xlabel('Importance')
-    ax.set_title(f'Top 20 Feature Importance (Epoch {epoch})')
-    ax.grid(axis='x', alpha=0.3)
-
-    buf = io.BytesIO()
-    plt.tight_layout()
-    plt.savefig(buf, format='png', dpi=100)
-    buf.seek(0)
-    img = Image.open(buf)
-    img_arr = np.array(img)
-    writer.add_image('FeatureImportance/Chart', img_arr, epoch, dataformats='HWC')
-    plt.close()
-
-def evaluate(perdicted_class, predicted_proba, true_class, writer: SummaryWriter, epoch: int):
-    accuracy = accuracy_score(true_class, perdicted_class)
-    precision = precision_score(true_class, perdicted_class, zero_division=0)
-    recall = recall_score(true_class, perdicted_class, zero_division=0)
-    f1 = f1_score(true_class, perdicted_class, zero_division=0)
-    roc_auc = roc_auc_score(true_class, predicted_proba)
-
-    writer.add_scalar('Eval/Accuracy', accuracy, epoch)
-    writer.add_scalar('Eval/Precision', precision, epoch)
-    writer.add_scalar('Eval/Recall', recall, epoch)
-    writer.add_scalar('Eval/F1_Score', f1, epoch)
-    writer.add_scalar('Eval/ROC_AUC', roc_auc, epoch)
-
-    # confusion matrix
-    cm = confusion_matrix(true_class, perdicted_class)
-    tn, fp, fn, tp = cm.ravel()
-    writer.add_scalar('Eval/TrueNeg', tn, epoch)
-    writer.add_scalar('Eval/FalsePos', fp, epoch)
-    writer.add_scalar('Eval/FalseNeg', fn, epoch)
-    writer.add_scalar('Eval/TruePos', tp, epoch)
-
-    # specificity and sensitivity
-    specificity = tn / (tn + fp) if (tn + fp) > 0 else 0
-    sensitivity = recall  # same as recall/TPR
-    writer.add_scalar('Eval/Specificity', specificity, epoch)
-    writer.add_scalar('Eval/Sensitivity', sensitivity, epoch)
-
-    fig, (ax1, ax2) = plt.subplots(1, 2, figsize=(12, 4))
-    ax1.matshow(cm, cmap='Blues', alpha=0.7)
-    for i in range(2):
-        for j in range(2):
-            ax1.text(j, i, str(cm[i, j]), ha='center', va='center', fontsize=14)
-    ax1.set_xlabel('Predicted')
-    ax1.set_ylabel('True')
-    ax1.set_title(f'Confusion Matrix (Epoch {epoch})')
-    ax1.set_xticks([0, 1])
-    ax1.set_yticks([0, 1])
-    ax1.set_xticklabels(['Human', 'Agent'])
-    ax1.set_yticklabels(['Human', 'Agent'])
-
-    # ROC curve
-    fpr, tpr, _ = roc_curve(true_class, predicted_proba)
-    ax2.plot(fpr, tpr, label=f'AUC={roc_auc:.3f}', linewidth=2)
-    ax2.plot([0, 1], [0, 1], 'k--', label='Random')
-    ax2.set_xlabel('False Positive Rate')
-    ax2.set_ylabel('True Positive Rate')
-    ax2.set_title('ROC Curve')
-    ax2.legend()
-    ax2.grid(alpha=0.3)
-
-    buf = io.BytesIO()
-    plt.tight_layout()
-    plt.savefig(buf, format='png', dpi=100)
-    buf.seek(0)
-    img = Image.open(buf)
-    img_arr = np.array(img)
-    writer.add_image('Eval/Metrics', img_arr, epoch, dataformats='HWC')
-    plt.close()
-
-    logger.info(f"Eval {epoch}: Acc={accuracy:.4f} Prec={precision:.4f} Rec={recall:.4f} F1={f1:.4f} AUC={roc_auc:.4f}")

experiments/ml/requirements.txt

@@ -1,6 +0,0 @@
-torch
-tensorboard
-fastparquet
-pyarrow
-xgboost
-lightgbm

experiments/ml/train.py

@@ -1,137 +0,0 @@
-from torch.utils.tensorboard import SummaryWriter
-from sklearn.model_selection import train_test_split
-from logging import getLogger
-from pathlib import Path
-import pandas as pd
-import numpy as np
-import joblib
-from datetime import datetime
-from ml.evals import evaluate, log_feature_importance
-from ml.arch import XGBoostAgentClassifier, LightGBMAgentClassifier, LABELS
-
-logger = getLogger(__name__)
-
-FEATURE_COLS_EXCLUDE = ['sessionId', 'experimentId', 'is_agent', 'xp_human_only', 'xp_market_mode', 'browser_family']
-RUNS_DIR = Path('ml/runs')
-CHECKPOINTS_DIR = Path('ml/checkpoints')
-
-
-def prepare_data(df):
-    """
-    Prepare feature matrix and labels from raw dataframe
-    Handles missing labels, feature selection, and categorical encoding
-    Returns: (X, y, feature_cols)
-    """
-    # drop rows with missing labels
-    n_before = len(df)
-    df = df[df['is_agent'].notna()].copy()
-    n_dropped = n_before - len(df)
-    if n_dropped > 0:
-        logger.warning(f"Dropped {n_dropped} sessions with missing labels")
-
-    if len(df) == 0:
-        logger.error("No labeled data available")
-        return None, None, None
-
-    feature_cols = [c for c in df.columns if c not in FEATURE_COLS_EXCLUDE]
-
-    # handle categorical browser_family via one-hot encoding
-    if 'browser_family' in df.columns:
-        browser_dummies = pd.get_dummies(df['browser_family'], prefix='browser', drop_first=True)
-        df = pd.concat([df, browser_dummies], axis=1)
-        feature_cols.extend(browser_dummies.columns.tolist())
-
-    X = df[feature_cols].fillna(0)
-    y = df['is_agent'].astype(int)
-
-    return X, y, feature_cols
-
-
-def train(data_path=None, model_type='xgboost', test_size=0.2, random_state=42,
-          n_estimators=200, max_depth=6, learning_rate=0.05):
-    """
-    Train agent detection classifier
-    Args:
-        data_path: path to labeled feature matrix CSV or parquet
-        model_type: 'xgboost' or 'lightgbm'
-        test_size: fraction for test split
-        random_state: seed for reproducibility
-    """
-    RUNS_DIR.mkdir(exist_ok=True)
-    CHECKPOINTS_DIR.mkdir(exist_ok=True)
-
-    run_name = f"{model_type}_{datetime.now().strftime('%Y%m%d_%H%M%S')}"
-    writer = SummaryWriter(log_dir=RUNS_DIR / run_name)
-    logger.info(f"Starting training run: {run_name}")
-
-    # load data
-    if data_path is None:
-        logger.error("data_path required")
-        return
-    df = pd.read_parquet(data_path)
-    logger.info(f"Loaded {len(df)} sessions from {data_path}")
-
-    # prepare features and labels
-    if 'is_agent' not in df.columns:
-        logger.error("Missing is_agent column")
-        return
-
-    X, y, feature_cols = prepare_data(df)
-    if X is None:
-        return
-
-    # class distribution
-    n_agents = y.sum()
-    n_humans = (y == 0).sum()
-    logger.info(f"Class distribution: {n_humans} humans, {n_agents} agents" + (f" (ratio {n_humans / n_agents:.2f})" if n_agents > 0 else ""))
-
-    # train/test split with stratification
-    X_train, X_test, y_train, y_test = train_test_split(
-        X, y, test_size=test_size, random_state=random_state, stratify=y
-    )
-    logger.info(f"Train: {len(X_train)}, Test: {len(X_test)}")
-
-    # init model
-    if model_type == 'xgboost':
-        model = XGBoostAgentClassifier(
-            n_estimators=n_estimators,
-            max_depth=max_depth,
-            learning_rate=learning_rate
-        )
-    elif model_type == 'lightgbm':
-        model = LightGBMAgentClassifier(
-            n_estimators=n_estimators,
-            max_depth=max_depth,
-            learning_rate=learning_rate
-        )
-    else:
-        logger.error(f"Unknown model type: {model_type}")
-        return
-
-    # train with eval set for early stopping
-    model.fit(X_train, y_train, eval_set=[(X_test, y_test)])
-    logger.info("Training complete")
-
-    # evaluate on test set
-    y_pred = model.predict(X_test)
-    y_prob = model.predict_proba(X_test)[:, 1]
-
-    evaluate(y_pred, y_prob, y_test, writer, epoch=0)
-
-    # log feature importance
-    log_feature_importance(writer, model, X.columns.tolist(), epoch=0)
-
-    # save model
-    model_path = CHECKPOINTS_DIR / f"{run_name}.pkl"
-    joblib.dump({'model': model, 'feature_cols': X.columns.tolist(), 'run_name': run_name}, model_path)
-    logger.info(f"Model saved to {model_path}")
-
-    writer.close()
-    return model, X.columns.tolist()
-
-
-if __name__ == "__main__":
-    import sys
-    data_path = sys.argv[1]
-    model_type = sys.argv[2] if len(sys.argv) > 2 else 'xgboost'
-    train(data_path, model_type=model_type)

experiments/procesing/__init__.py

@@ -12,14 +12,16 @@ from procesing.steps import (
     ComputeDemandStep,
     ComputeDemandForChunksStep,
     AggregatePriceLogsStep,
-    # StateSpace,
-    # BuildStateSpaceStep,
+    ComputeElasticityStep,
+    StateSpace,
+    BuildStateSpaceStep,
     FitPricingFunctionStep,
     PredictPricesStep,
 )
 from procesing.pipelines import (
     interaction_extraction_pipeline,
     price_extraction_pipeline,
+    elasticity_computation_pipeline,
     pricing_pipeline,
     full_pipeline,
 )
@@ -40,12 +42,14 @@ __all__ = [
     'ComputeDemandStep',
     'ComputeDemandForChunksStep',
     'AggregatePriceLogsStep',
-    # 'StateSpace',
-    # 'BuildStateSpaceStep',
+    'ComputeElasticityStep',
+    'StateSpace',
+    'BuildStateSpaceStep',
     'FitPricingFunctionStep',
     'PredictPricesStep',
     'interaction_extraction_pipeline',
     'price_extraction_pipeline',
+    'elasticity_computation_pipeline',
     'pricing_pipeline',
     'full_pipeline',
 ]

experiments/procesing/pipelines.py

@@ -2,7 +2,7 @@ from sklearn.pipeline import Pipeline
 import pandas as pd
 from procesing.context import PipelineContext
 from procesing.providers import SupabaseProvider, BackendAPIProvider
-import os
+from typing import Union
 from procesing.steps import (
     FetchInteractionsStep,
     FetchPriceLogsStep,
@@ -13,15 +13,11 @@ from procesing.steps import (
     ChunkByTimeWindowStep,
     ComputeDemandForChunksStep,
     AggregatePriceLogsStep,
+    ComputeElasticityStep,
+    BuildStateSpaceStep,
     FitPricingFunctionStep,
     PredictPricesStep,
-    ComputeDemandStep,
-    JoinProductFeaturesStep,
-    ExtractSessionFeaturesStep,
-    JoinLabelsStep,
-    ValidateDataStep,
 )
-from procesing.pricers import SimpleSurgePricer
 
 def interaction_extraction_pipeline(context: PipelineContext):
     """Pipeline for extracting and augmenting interaction data"""
@@ -39,136 +35,104 @@ def price_extraction_pipeline(context: PipelineContext):
     ])
 
 
-def product_features_pipeline(context: PipelineContext,
+def elasticity_computation_pipeline(context: PipelineContext,
                                     interactions_df: pd.DataFrame,
                                     price_logs_df: pd.DataFrame):
-    demand_step = ComputeDemandStep(context)
+    """
+    Compute elasticity from interactions and price logs.
+    Manual orchestration needed for branching logic.
+    """
+    # branch 1: chunk interactions and compute demand
+    chunk_step = ChunkByTimeWindowStep(context)
+    interaction_chunks = chunk_step.transform(interactions_df)
+
+    demand_step = ComputeDemandForChunksStep(context)
+    demand_chunks = demand_step.transform(interaction_chunks)
+
+    # branch 2: aggregate price logs
     price_step = AggregatePriceLogsStep(context)
-    join_step = JoinProductFeaturesStep(context)
+    price_chunks = price_step.transform(price_logs_df)
+
+    # convergence: compute elasticity
+    elasticity_step = ComputeElasticityStep(context)
+    elasticity_df = elasticity_step.transform((demand_chunks, price_chunks))
+
+    return elasticity_df
 
 
-    demand_data = demand_step.transform(interactions_df)
-    price_data= price_step.transform(price_logs_df)
-    joined_data = join_step.transform((demand_data, price_data))
-
-    return joined_data
-
-
-
-def pricing_pipeline(context: "PipelineContext",
-                     data: pd.DataFrame,
-                     high_threshold: int = 10,
-                     low_threshold: int = 2,
-                     surge_multiplier: float = 1.2,
-                     discount_multiplier: float = 0.9) -> pd.DataFrame:
-
-    if data.empty or 'productId' not in data.columns:
-        return pd.DataFrame()
-
-    surge_pricer = SimpleSurgePricer()
-    surge_pricer.fit(data)
-    data['optimal_price'] = surge_pricer.predict()
-    return data
-
-
-def full_pipeline(context: PipelineContext,
-                  high_threshold: int = 10,
-                  low_threshold: int = 2,
-                  surge_multiplier: float = 1.2,
-                  discount_multiplier: float = 0.9):
+def pricing_pipeline(context: PipelineContext, elasticity_df: pd.DataFrame):
     """
-    Complete end-to-end pipeline: data extraction -> demand/price aggregation -> surge pricing
-
-    Args:
-        context: Pipeline context
-        high_threshold: Demand threshold for surge pricing
-        low_threshold: Demand threshold for discounts
-        surge_multiplier: Price multiplier for high demand
-        discount_multiplier: Price multiplier for low demand
-
-    Returns:
-        tuple: (product_features_df, optimal_prices_df)
-            - product_features_df: [productId, demand_score, price]
-            - optimal_prices_df: [productId, current_price, optimal_price, demand_score]
+    Generate optimal prices from elasticity estimates.
     """
+    # build state space
+    state_step = BuildStateSpaceStep(context)
+    state_space = state_step.transform(elasticity_df)
+
+    # fit pricing function
+    fit_step = FitPricingFunctionStep(context)
+    pricer = fit_step.transform(elasticity_df)
+
+    # predict prices
+    predict_step = PredictPricesStep(context)
+    prices_df = predict_step.transform((pricer, state_space))
+
+    return prices_df
+
+
+def full_pipeline(context: PipelineContext):
+    """
+    Complete end-to-end pipeline: data extraction -> elasticity -> pricing
+    Returns: (elasticity_df, prices_df)
+    """
+    # extract interactions
     interaction_pipe = interaction_extraction_pipeline(context)
-    price_pipe = price_extraction_pipeline(context)
-
     interactions_df = interaction_pipe.fit_transform(None)
+
+    # extract price logs
+    price_pipe = price_extraction_pipeline(context)
     price_logs_df = price_pipe.fit_transform(None)
-    product_features_df = product_features_pipeline(context, interactions_df, price_logs_df)
-    print(product_features_df.to_string())
 
-    # generate optimal prices using surge rules
-    optimal_prices_df = pricing_pipeline(context, product_features_df,
-                                          high_threshold=high_threshold,
-                                          low_threshold=low_threshold,
-                                          surge_multiplier=surge_multiplier,
-                                          discount_multiplier=discount_multiplier)
+    if interactions_df.empty or price_logs_df.empty:
+        return None, None
 
-    return product_features_df, optimal_prices_df
+    # compute elasticity
+    elasticity_df = elasticity_computation_pipeline(
+        context,
+        interactions_df,
+        price_logs_df
+    )
 
+    if elasticity_df is None or elasticity_df.empty:
+        return elasticity_df, None
 
-def ml_training_pipeline(context: PipelineContext) -> pd.DataFrame:
-    """
-    Build labeled session-level feature matrix for ML model training.
-    Pipeline: fetch -> validate -> extract features -> join labels
-
-    Returns:
-        DataFrame with ~25 features per session + is_agent label
-        Columns: sessionId, experimentId, temporal/behavioral/product/ua features, is_agent
-    """
-    # fetch raw interactions
-    interactions_df = FetchInteractionsStep(context).transform(None)
-
-    # validate data quality (report cached in context)
-    interactions_df = ValidateDataStep(context).transform(interactions_df)
-    if interactions_df.empty:
-        return pd.DataFrame()
-
-    # extract vectorized session features
-    features_df = ExtractSessionFeaturesStep(context).transform(interactions_df)
-    if features_df.empty:
-        return pd.DataFrame()
-
-    # join experiment labels (is_agent = ~xp_human_only)
-    labeled_df = JoinLabelsStep(context).transform(features_df)
-
-    return labeled_df
-
+    # generate prices
+    prices_df = pricing_pipeline(context, elasticity_df)
 
+    return elasticity_df, prices_df
 
 
 if __name__ == '__main__':
 
-    class ExperimentsProvider(SupabaseProvider, BackendAPIProvider):
-        def fetch_kafka_topic(self, topic: str) -> pd.DataFrame:
-            base_path = "/home/velocitatem/Documents/Projects/PHANTOM/experiments/collected_data/" # os.path.join(os.path.dirname(__file__), "collected_data")
-            if not os.path.isdir(base_path):
-                return pd.DataFrame()
+    class Provider(SupabaseProvider, BackendAPIProvider):
+        def __init__(self, backend_url: str):
+            SupabaseProvider.__init__(self)
+            BackendAPIProvider.__init__(self, backend_url=backend_url)
+    # example run
+    context = PipelineContext(
+        provider=Provider(backend_url="http://localhost:5000"),
+        store_mode='hotel',
+    )
 
-            files = {"user-interactions": "int.json", "price-logs": "price.json"}
-            file_to_read = files.get(topic, files["user-interactions"])
-            frames = []
+    elasticity_df, prices_df = full_pipeline(context)
 
-            for d in os.listdir(base_path):
-                full_path = os.path.join(base_path, d, file_to_read)
-                if not os.path.isfile(full_path):
-                    continue
-                try:
-                    data = pd.read_json(full_path)
-                    payloads = pd.DataFrame([r['payload'] for r in data['value'].to_list()])
-                    frames.append(payloads)
-                except Exception as e:
-                    print(f"Warning: Could not process {full_path}: {e}")
+    if elasticity_df is not None and not elasticity_df.empty:
+        print("Elasticity Estimates:")
+        print(elasticity_df.to_string(index=False))
+    else:
+        print("No elasticity estimates computed.")
 
-            return pd.concat(frames, ignore_index=True) if frames else pd.DataFrame()
-
-    # demo: run ML training pipeline
-    context = PipelineContext(provider=ExperimentsProvider(), store_mode='hotel')
-    features = ml_training_pipeline(context)
-    print(f"Feature matrix: {features.shape}")
-    print(features.head())
-    print(features.info())
-
-    features.to_parquet("features.parquet")
+    if prices_df is not None and not prices_df.empty:
+        print("\nPredicted Prices:")
+        print(prices_df.to_string(index=False))
+    else:
+        print("No prices predicted.")

experiments/procesing/pricers/__init__.py

@@ -1,6 +1,6 @@
 from procesing.pricers.base import PricingFunction
 from procesing.pricers.elasticity import ElasticityBasedPricer
-from procesing.pricers.simple import StaticPricer, RandomPricer, SimpleSurgePricer
+from procesing.pricers.simple import StaticPricer, RandomPricer
 from procesing.pricers.session_aware import SessionAwarePricer, ProductSpecificSessionPricer
 
 __all__ = [
@@ -8,7 +8,6 @@ __all__ = [
     'ElasticityBasedPricer',
     'StaticPricer',
     'RandomPricer',
-    'SimpleSurgePricer',
     'SessionAwarePricer',
     'ProductSpecificSessionPricer'
 ]

experiments/procesing/pricers/base.py

@@ -25,7 +25,7 @@ class PricingFunction(ABC):
     """
 
     @abstractmethod
-    def fit(self, *kwargs):
+    def fit(self, historical_data: pd.DataFrame, **kwargs):
         """
         Offline training on historical data.
 
@@ -36,7 +36,7 @@ class PricingFunction(ABC):
         pass
 
     @abstractmethod
-    def predict(self, *kwargs) -> np.ndarray:
+    def predict(self, state_space) -> np.ndarray:
         """
         Generate optimal prices given current state.
 

experiments/procesing/pricers/simple.py

@@ -46,46 +46,3 @@ class RandomPricer(PricingFunction):
         if self.n_products is None:
             self.n_products = len(state_space.demand)
         return self.rng.uniform(self.price_min, self.price_max, size=self.n_products)
-
-
-class SimpleSurgePricer(PricingFunction):
-    """
-    Rule-based surge pricer adjusting prices via demand thresholds.
-    Logic: if demand > high_threshold -> surge, if demand < low_threshold -> discount.
-    Simpler and more controllable than curve fitting approaches.
-    """
-
-    def __init__(self,
-                 base_prices: np.ndarray = None,
-                 high_threshold: int = 10,
-                 low_threshold: int = 2,
-                 surge_multiplier: float = 1.2,
-                 discount_multiplier: float = 0.9):
-        self.base_prices = base_prices
-        self.high_threshold = high_threshold
-        self.low_threshold = low_threshold
-        self.surge_multiplier = surge_multiplier
-        self.discount_multiplier = discount_multiplier
-
-    def fit(self, market_data : pd.DataFrame):
-        """Extract base prices from product catalog or historical averages"""
-        self.base_prices = market_data['base_price'].to_numpy() if 'base_price' in market_data.columns else market_data['price'].values
-        self.demand_history = market_data['demand'].to_numpy() if 'demand' in market_data.columns else np.zeros_like(self.base_prices)
-
-    def predict(self) -> np.ndarray:
-        """
-        Adjust prices based on current demand using surge rules.
-        state_space.demand: demand counts per product
-        state_space.prices: current prices (fallback if base_prices not set)
-        """
-        current_prices = self.base_prices if self.base_prices is not None else np.ones_like(demand_vector) * 99.99
-        demand = self.demand_history if self.demand_history is not None else np.zeros_like(current_prices)
-        new_prices = current_prices.copy()
-
-        high_mask = demand >= self.high_threshold
-        new_prices[high_mask] *= self.surge_multiplier
-
-        low_mask = demand <= self.low_threshold
-        new_prices[low_mask] *= self.discount_multiplier
-
-        return new_prices

experiments/procesing/providers/supabase.py

@@ -18,17 +18,10 @@ class SupabaseProvider(DataProvider):
         self.supabase: Client = create_client(self.supabase_url, self.supabase_key)
 
     def fetch_products(self, store_mode: str) -> pd.DataFrame:
-        # hotel uses room_type, airline uses flight_type; select all and normalize
-        resp = self.supabase.table(f'{store_mode}_products').select("*").execute()
-        if not resp.data:
-            return pd.DataFrame()
-        df = pd.DataFrame(resp.data)
-        # normalize type column: hotel has room_type, airline has flight_type
-        if 'room_type' in df.columns:
-            df['product_type'] = df['room_type']
-        elif 'flight_type' in df.columns:
-            df['product_type'] = df['flight_type']
-        return df
+        resp = self.supabase.table(f'{store_mode}_products').select(
+            "id, room_type, date_index, metadata, availability"
+        ).execute()
+        return pd.DataFrame(resp.data) if resp.data else pd.DataFrame()
 
     def fetch_experiments(self, experiment_ids: List[str]) -> pd.DataFrame:
         if not experiment_ids:

experiments/procesing/steps/__init__.py

@@ -1,16 +1,11 @@
 from procesing.steps.base import BaseContextStep
 from procesing.steps.fetch import FetchInteractionsStep, FetchPriceLogsStep, FetchExperimentsStep
-from procesing.steps.join import JoinExperimentsStep, JoinProductFeaturesStep
-from procesing.steps.augment import CreatePriceBucketsStep, AugmentEventNamesStep, AugmentInteractionsStep
+from procesing.steps.join import JoinExperimentsStep
+from procesing.steps.augment import CreatePriceBucketsStep, AugmentEventNamesStep
 from procesing.steps.chunk import ChunkByTimeWindowStep
 from procesing.steps.demand import ComputeDemandStep, ComputeDemandForChunksStep
-from procesing.steps.elasticity import AggregatePriceLogsStep
-from procesing.steps.pricing import FitPricingFunctionStep, PredictPricesStep
-from procesing.steps.session import (
-    ExtractSessionFeaturesStep, JoinLabelsStep, ValidateDataStep,
-    TemporalFeatureStep, BehavioralFeatureStep, ProductFeatureStep, UserAgentFeatureStep,
-    _extract_features_for_session
-)
+from procesing.steps.elasticity import AggregatePriceLogsStep, ComputeElasticityStep
+from procesing.steps.pricing import StateSpace, BuildStateSpaceStep, FitPricingFunctionStep, PredictPricesStep
 
 __all__ = [
     'BaseContextStep',
@@ -18,22 +13,15 @@ __all__ = [
     'FetchPriceLogsStep',
     'FetchExperimentsStep',
     'JoinExperimentsStep',
-    'JoinProductFeaturesStep',
     'CreatePriceBucketsStep',
     'AugmentEventNamesStep',
-    'AugmentInteractionsStep',
     'ChunkByTimeWindowStep',
     'ComputeDemandStep',
     'ComputeDemandForChunksStep',
     'AggregatePriceLogsStep',
+    'ComputeElasticityStep',
+    'StateSpace',
+    'BuildStateSpaceStep',
     'FitPricingFunctionStep',
     'PredictPricesStep',
-    'ExtractSessionFeaturesStep',
-    'JoinLabelsStep',
-    'ValidateDataStep',
-    'TemporalFeatureStep',
-    'BehavioralFeatureStep',
-    'ProductFeatureStep',
-    'UserAgentFeatureStep',
-    '_extract_features_for_session',
 ]

experiments/procesing/steps/augment.py

@@ -2,93 +2,6 @@ import numpy as np
 import pandas as pd
 from procesing.steps.base import BaseContextStep
 
-
-class AugmentInteractionsStep(BaseContextStep):
-    """
-    Consolidated step: create price buckets, augment event names, join experiments.
-    Input: (interactions_df, price_logs_df)
-    Output: enriched interactions_df
-    """
-
-    def transform(self, data: tuple):
-        interactions_df, price_logs_df = data
-
-        if interactions_df.empty:
-            return interactions_df
-
-        # Step 1: Create price buckets
-        interactions_df = self._create_price_buckets(interactions_df)
-
-        # Step 2: Augment event names
-        interactions_df = self._augment_event_names(interactions_df)
-
-        # Step 3: Join experiments (optional)
-        if 'experimentId' in interactions_df.columns:
-            interactions_df = self._join_experiments(interactions_df)
-
-        return interactions_df
-
-    def _create_price_buckets(self, df: pd.DataFrame):
-        """Create price bucket labels from price data"""
-        if 'metadata_price' not in df.columns:
-            df['price_bucket'] = ""
-            return df
-
-        n_buckets = self.context.config.get('n_price_buckets', 5)
-
-        if df['metadata_price'].notnull().sum() > 0:
-            try:
-                price_buckets = pd.qcut(
-                    df['metadata_price'],
-                    q=n_buckets,
-                    labels=[f"PB_{i+1}" for i in range(n_buckets)],
-                    duplicates='drop'
-                )
-            except ValueError:
-                # fallback for insufficient unique values
-                price_buckets = df['metadata_price'].apply(
-                    lambda x: f"P_{int(x)}" if pd.notnull(x) else ""
-                )
-        else:
-            price_buckets = pd.Series([""] * len(df), index=df.index)
-
-        df['price_bucket'] = price_buckets
-        return df
-
-    def _augment_event_names(self, df: pd.DataFrame):
-        """Augment event names with product and price bucket schema"""
-        # Create schema: _productId@price_bucket
-        has_product = df.get('productId', pd.Series()).notnull()
-        has_bucket = df.get('price_bucket', pd.Series()).notnull()
-
-        df['metadata_schema'] = np.where(
-            has_product & has_bucket,
-            "_" + df['productId'].astype(str) + "@" + df['price_bucket'].astype(str),
-            ""
-        )
-
-        df['eventName'] = df['eventName'] + df['metadata_schema']
-        return df
-
-    def _join_experiments(self, df: pd.DataFrame):
-        """Join experiment metadata if experimentId present"""
-        exp_ids = df['experimentId'].dropna().unique().tolist()
-        if not exp_ids:
-            return df
-
-        experiments_df = self.context.provider.fetch_experiments(exp_ids)
-        if experiments_df.empty:
-            return df
-
-        return df.merge(
-            experiments_df,
-            left_on='experimentId',
-            right_on='id',
-            how='left',
-            suffixes=('', '_exp')
-        )
-
-
 class CreatePriceBucketsStep(BaseContextStep):
     """Create price bucket labels from price data"""
 

experiments/procesing/steps/base.py

@@ -1,7 +1,6 @@
 from abc import ABC, abstractmethod
 from sklearn.base import BaseEstimator, TransformerMixin
 from procesing.context import PipelineContext
-from typing import Any
 
 class BaseContextStep(BaseEstimator, TransformerMixin, ABC):
     """
@@ -17,7 +16,7 @@ class BaseContextStep(BaseEstimator, TransformerMixin, ABC):
         return self
 
     @abstractmethod
-    def transform(self, X) -> Any:
+    def transform(self, X):
         """Transform input using context. Must be implemented by subclass."""
         pass
 

experiments/procesing/steps/elasticity.py

@@ -7,16 +7,16 @@ class AggregatePriceLogsStep(BaseContextStep):
     """
     Aggregate price logs into time windows using VECTORIZED operations.
     Input: price_logs_df
-    Output: DataFrame with columns [productId, price]
+    Output: list of price chunks with [productId, price]
     """
 
     def transform(self, price_logs_df: pd.DataFrame):
         if price_logs_df.empty:
-            return pd.DataFrame(columns=['productId', 'price'])
+            return []
 
         df = price_logs_df.copy()
         ts_col = self.context.config.get('ts_col', 'ts')
-        #window_size = self.context.window_size WE ARE NOT USING CHUNKS ANYMORE
+        window_size = self.context.window_size
 
         # ensure datetime
         if not pd.api.types.is_datetime64_any_dtype(df[ts_col]):
@@ -24,19 +24,230 @@ class AggregatePriceLogsStep(BaseContextStep):
 
         df = df.sort_values([ts_col, 'productId'])
         products = self.context.products
-        # get base price from metadata if available 1) read the metadata col as json and get the base_price
-        products['base_price'] = products.apply(
-            lambda row: row['metadata'].get('base_price', 0) if isinstance(row['metadata'], dict) else 0,
-            axis=1
-        )
-
         unique_products = products['id'].unique()
 
+        # VECTORIZED: group by product, resample by time window, compute mean
         df_indexed = df.set_index(ts_col)
-        # we return a df of average price per product over the entire period
-        # TODO: maybe consider different opration to handle price aggregation over time
-        avg_prices = df_indexed.groupby('productId')['price'].mean().reindex(unique_products, fill_value=0).reset_index()
-        avg_prices.columns = ['productId', 'price']
-        # fill 0s with base_price from products
-        base_price_map = products.set_index('id')['base_price'].to_dict()
-        return avg_prices
+
+        windowed = (
+            df_indexed
+            .groupby('productId')['price']
+            .resample(window_size)
+            .mean()
+            .reset_index()
+        )
+
+        # forward fill missing windows (carry last known price)
+        windowed = windowed.sort_values([ts_col, 'productId'])
+        windowed['price'] = windowed.groupby('productId')['price'].ffill()
+        windowed = windowed.dropna(subset=['price'])
+
+        # group into chunks by window
+        chunks = []
+        for window_start, group in windowed.groupby(ts_col):
+            price_vector = group[['productId', 'price']].copy()
+
+            # fill missing products with last known price before this window
+            missing_products = set(unique_products) - set(price_vector['productId'])
+            if missing_products:
+                for pid in missing_products:
+                    last_price = df_indexed[
+                        (df_indexed['productId'] == pid) &
+                        (df_indexed.index < window_start)
+                    ]['price']
+
+                    if not last_price.empty:
+                        price_vector = pd.concat([
+                            price_vector,
+                            pd.DataFrame({'productId': [pid], 'price': [last_price.iloc[-1]]})
+                        ], ignore_index=True)
+
+            if not price_vector.empty:
+                chunks.append({
+                    'window_start': window_start,
+                    'window_end': window_start + pd.Timedelta(window_size),
+                    'price_vector': price_vector
+                })
+
+        return chunks
+
+
+class ComputeElasticityStep(BaseContextStep):
+    """
+    Compute price elasticity from demand and price chunks.
+    Input: (demand_chunks, price_chunks)
+    Output: elasticity_df [productId, elasticity, std_error, n_obs]
+    """
+
+    def transform(self, chunk_tuple: tuple):
+        demand_chunks, price_chunks = chunk_tuple
+
+        method = self.context.config.get('elasticity_method', 'point')
+        min_obs = self.context.config.get('min_observations', 2)
+
+        products = self.context.products
+        all_product_ids = products['id'].unique()
+
+        # align chunks by window_start
+        aligned = self._align_chunks(demand_chunks, price_chunks)
+
+        if not aligned:
+            return pd.DataFrame({
+                'productId': all_product_ids,
+                'elasticity': 0.0,
+                'std_error': 0.0,
+                'n_obs': 0
+            })
+
+        # build time series per product
+        product_series = self._build_timeseries(aligned)
+
+        # compute elasticity per product
+        elasticities = []
+        for pid, series in product_series.items():
+            if len(series) < min_obs:
+                elasticities.append({
+                    'productId': pid,
+                    'elasticity': 0.0,
+                    'std_error': 0.0,
+                    'n_obs': len(series)
+                })
+                continue
+
+            elast = self._compute_elasticity(series, method)
+            elasticities.append({
+                'productId': pid,
+                'elasticity': elast['value'],
+                'std_error': elast.get('std_error', 0.0),
+                'n_obs': len(series)
+            })
+
+        result_df = pd.DataFrame(elasticities)
+
+        # fill missing products with zero elasticity
+        observed_pids = set(result_df['productId'])
+        missing_pids = [p for p in all_product_ids if p not in observed_pids]
+
+        if missing_pids:
+            missing_df = pd.DataFrame({
+                'productId': missing_pids,
+                'elasticity': 0.0,
+                'std_error': 0.0,
+                'n_obs': 0
+            })
+            result_df = pd.concat([result_df, missing_df], ignore_index=True)
+
+        return result_df
+
+    def _align_chunks(self, demand_chunks: List[Dict], price_chunks: List[Dict]):
+        """Align demand and price chunks by window_start"""
+        price_lookup = {c['window_start']: c for c in price_chunks}
+        aligned = []
+
+        for dc in demand_chunks:
+            ws = dc['window_start']
+            if ws in price_lookup:
+                aligned.append({
+                    'window_start': ws,
+                    'window_end': dc['window_end'],
+                    'demand': dc['demand_vector'],
+                    'prices': price_lookup[ws]['price_vector']
+                })
+
+        return aligned
+
+    def _build_timeseries(self, aligned: List[Dict]):
+        """Build time series [timestamp, price, quantity] per product"""
+        series_by_product = {}
+
+        for chunk in aligned:
+            merged = chunk['demand'].merge(chunk['prices'], on='productId', how='inner')
+
+            for _, row in merged.iterrows():
+                pid = row['productId']
+                if pid not in series_by_product:
+                    series_by_product[pid] = []
+
+                series_by_product[pid].append({
+                    'timestamp': chunk['window_start'],
+                    'price': row['price'],
+                    'quantity': row['demand_score']
+                })
+
+        return series_by_product
+
+    def _compute_elasticity(self, series: List[Dict], method: str):
+        """Compute point or arc elasticity"""
+        prices = np.array([s['price'] for s in series])
+        quantities = np.array([s['quantity'] for s in series])
+
+        # filter out zero/negative values
+        valid = (prices > 0) & (quantities > 0)
+        if valid.sum() < 2:
+            return {'value': 0.0, 'std_error': 0.0}
+
+        prices = prices[valid]
+        quantities = quantities[valid]
+
+        if method == 'point':
+            return self._point_elasticity(prices, quantities)
+        elif method == 'arc':
+            return self._arc_elasticity(prices, quantities)
+        else:
+            raise ValueError(f"Unknown elasticity method: {method}")
+
+    def _point_elasticity(self, prices: np.ndarray, quantities: np.ndarray):
+        """Point elasticity via log-log regression: log(Q) = a + b*log(P), elasticity = b"""
+        if len(prices) < 2:
+            return {'value': 0.0, 'std_error': 0.0}
+
+        log_p = np.log(prices)
+        log_q = np.log(quantities)
+
+        if log_p.std() == 0:
+            return {'value': 0.0, 'std_error': 0.0}
+
+        cov = np.cov(log_p, log_q)[0, 1]
+        var = np.var(log_p)
+        b = cov / var
+
+        # std error estimate
+        if len(prices) > 2:
+            residuals = log_q - (log_q.mean() + b * (log_p - log_p.mean()))
+            mse = (residuals ** 2).sum() / (len(prices) - 2)
+            se_b = np.sqrt(mse / (len(prices) * var))
+        else:
+            se_b = 0.0
+
+        return {'value': b, 'std_error': se_b}
+
+    def _arc_elasticity(self, prices: np.ndarray, quantities: np.ndarray):
+        """Arc elasticity: average period-over-period elasticity"""
+        elasticities = []
+
+        for i in range(1, len(prices)):
+            p1, p2 = prices[i-1], prices[i]
+            q1, q2 = quantities[i-1], quantities[i]
+
+            p_avg = (p1 + p2) / 2
+            q_avg = (q1 + q2) / 2
+
+            if p_avg == 0 or q_avg == 0:
+                continue
+
+            delta_p = p2 - p1
+            delta_q = q2 - q1
+
+            if delta_p == 0:
+                continue
+
+            e = (delta_q / q_avg) / (delta_p / p_avg)
+            elasticities.append(e)
+
+        if not elasticities:
+            return {'value': 0.0, 'std_error': 0.0}
+
+        return {
+            'value': np.mean(elasticities),
+            'std_error': np.std(elasticities) / np.sqrt(len(elasticities))
+        }

experiments/procesing/steps/fetch.py

@@ -2,11 +2,7 @@ import pandas as pd
 from procesing.steps.base import BaseContextStep
 
 class FetchInteractionsStep(BaseContextStep):
-    """Fetch raw interaction data from Kafka topic with optional time and store_mode filtering"""
-
-    def __init__(self, context, lookback: str = None):
-        super().__init__(context)
-        self.lookback = lookback
+    """Fetch raw interaction data from Kafka topic"""
 
     def transform(self, X=None):
         df = self.context.provider.fetch_kafka_topic('user-interactions')
@@ -21,50 +17,19 @@ class FetchInteractionsStep(BaseContextStep):
             )
 
         df = df.dropna(subset=['eventName'])
-        # drop all where page has /admin/
-        df = df[~df['page'].str.contains('/admin/', na=False)]
-
-        # filter by store_mode from context
-        if 'storeMode' in df.columns:
-            df = df[df['storeMode'] == self.context.store_mode]
 
         # Remap dateIndex if present
         if 'metadata_dateIndex' in df.columns:
             df['dateIndex'] = df['metadata_dateIndex'].astype('Int64')
 
-        # Apply time filtering if lookback specified
-        if self.lookback and 'ts' in df.columns:
-            df['ts'] = pd.to_datetime(df['ts'])
-            cutoff = pd.Timestamp.now() - pd.Timedelta(self.lookback)
-            df = df[df['ts'] >= cutoff]
-
         return df
 
 
 class FetchPriceLogsStep(BaseContextStep):
-    """Fetch price log data from Kafka topic with optional time and store_mode filtering"""
-
-    def __init__(self, context, lookback: str = None):
-        super().__init__(context)
-        self.lookback = lookback
+    """Fetch price log data from Kafka topic"""
 
     def transform(self, X=None):
-        df = self.context.provider.fetch_kafka_topic('price-logs')
-
-        if df.empty:
-            return df
-
-        # filter by store_mode from context
-        if 'storeMode' in df.columns:
-            df = df[df['storeMode'] == self.context.store_mode]
-
-        # Apply time filtering if lookback specified
-        if self.lookback and 'ts' in df.columns:
-            df['ts'] = pd.to_datetime(df['ts'])
-            cutoff = pd.Timestamp.now() - pd.Timedelta(self.lookback)
-            df = df[df['ts'] >= cutoff]
-
-        return df
+        return self.context.provider.fetch_kafka_topic('price-logs')
 
 
 class FetchExperimentsStep(BaseContextStep):

experiments/procesing/steps/join.py

@@ -32,27 +32,3 @@ class JoinExperimentsStep(BaseContextStep):
         })
 
         return interactions_df.merge(experiments_df, on='experimentId', how='left')
-
-class JoinProductFeaturesStep(BaseContextStep):
-    """Join product features to interactions"""
-
-    def transform(self, data: tuple):
-        """
-        Args:
-            data: (interactions_df, products_df)
-        Returns:
-            merged interactions dataframe
-        """
-        demand_df, price_df = data
-
-        # get base prices from products if available
-        products = self.context.products
-        products['base_price'] = products.apply(
-            lambda row: float(row['metadata'].get('base_price', 0.0)) if isinstance(row['metadata'], dict) else 0,
-            axis=1
-        )
-        products = products[['id', 'base_price']].rename(columns={'id': 'productId'})
-
-        if price_df.empty:
-            return demand_df
-        return demand_df.merge(price_df, on='productId', how='left').merge(products, on='productId', how='left')

experiments/procesing/steps/pricing.py

@@ -2,34 +2,128 @@ import numpy as np
 import pandas as pd
 from typing import Optional, List, Dict, Any
 from dataclasses import dataclass, field
-from procesing.pricers.simple import StaticPricer
 from procesing.steps.base import BaseContextStep
 from procesing.pricers import ElasticityBasedPricer
 
-class State:
-    def __init__(self,
-                 last_action : str,
-                 last_productId : str,
-                 last_price : float,
-                 session_features : np.ndarray
-                 ):
-        pass
+@dataclass
+class StateSpace:
+    """
+    State representation for pricing functions.
 
+    Components:
+        Q_t: demand ∈ R^n (current demand signal per product)
+        P_t: prices ∈ R^n (current/base prices)
+        S_t: session_features (behavioral signals, interaction data)
+        H_t: history = {Q_{t-k}, P_{t-k}, S_{t-k}} for k in [1, history_length]
+
+    Additionally stores:
+        - product_ids: product identifiers (n,)
+        - elasticity: price elasticity per product (n,)
+        - metadata: arbitrary context (experiment_id, timestamp, etc.)
+    """
+    demand: np.ndarray  # Q_t ∈ R^n
+    prices: np.ndarray  # P_t ∈ R^n
+    session_features: pd.DataFrame = field(default_factory=pd.DataFrame)  # S_t
+
+    # augmented state components
+    product_ids: Optional[np.ndarray] = None
+    elasticity: Optional[np.ndarray] = None
+
+    # historical trajectory H_t = {(Q_{t-k}, P_{t-k}, S_{t-k})}
+    history: List[Dict[str, Any]] = field(default_factory=list)
+
+    # metadata for context
+    metadata: Dict[str, Any] = field(default_factory=dict)
+
+    def __post_init__(self):
+        """Validate dimensions."""
+        n = len(self.demand)
+        assert len(self.prices) == n, "demand and prices must have same dimension"
+        if self.elasticity is not None:
+            assert len(self.elasticity) == n, "elasticity must match dimension"
+        if self.product_ids is not None:
+            assert len(self.product_ids) == n, "product_ids must match dimension"
+
+    @property
+    def n_products(self) -> int:
+        """Number of products in state space."""
+        return len(self.demand)
+
+    def add_history(self, q: np.ndarray, p: np.ndarray, s: pd.DataFrame, max_length: int = 10):
+        """Append historical state to trajectory H_t."""
+        self.history.append({'demand': q, 'prices': p, 'session_features': s})
+        if len(self.history) > max_length:
+            self.history.pop(0)
+
+    def get_history_window(self, k: int = 5) -> List[Dict[str, Any]]:
+        """Retrieve last k historical states."""
+        return self.history[-k:] if len(self.history) >= k else self.history
+
+
+class BuildStateSpaceStep(BaseContextStep):
+    """
+    Build state space from elasticity, demand, and price data.
+
+    Input: elasticity_df [productId, elasticity, ...], optional demand_df
+    Output: StateSpace instance with Q_t, P_t, elasticity, product_ids
+    """
+
+    def transform(self, elasticity_df: pd.DataFrame, demand_df: Optional[pd.DataFrame] = None):
+        products = self.context.products
+
+        # extract base prices from product metadata
+        products_with_prices = products.copy()
+        if 'metadata' in products_with_prices.columns:
+            products_with_prices['base_price'] = products_with_prices['metadata'].apply(
+                lambda m: m.get('base_price', 0) if isinstance(m, dict) else 0
+            )
+        else:
+            products_with_prices['base_price'] = 0
+
+        # merge with elasticity
+        merged = products_with_prices[['id', 'base_price']].rename(
+            columns={'id': 'productId'}
+        ).merge(
+            elasticity_df[['productId', 'elasticity']],
+            on='productId',
+            how='left'
+        ).fillna({'elasticity': 0.0, 'base_price': 0.0})
+
+        # merge with demand if provided, else use default
+        if demand_df is not None and 'demand' in demand_df.columns:
+            merged = merged.merge(
+                demand_df[['productId', 'demand']],
+                on='productId',
+                how='left'
+            ).fillna({'demand': 0.0})
+            demand_vector = merged['demand'].values
+        else:
+            # default: uniform demand or use elasticity as proxy
+            demand_vector = np.ones(len(merged)) * 10.0
+
+        return StateSpace(
+            demand=demand_vector,
+            prices=merged['base_price'].values,
+            session_features=pd.DataFrame(),
+            product_ids=merged['productId'].values,
+            elasticity=merged['elasticity'].values,
+            metadata={'timestamp': pd.Timestamp.now().isoformat()}
+        )
 
 
 class FitPricingFunctionStep(BaseContextStep):
     """
-    Fit pricing function using data.
-    Input: pricing_data
+    Fit pricing function using elasticity data.
+    Input: elasticity_df
     Output: fitted pricing function instance
     """
 
-    def transform(self, pricing_data: pd.DataFrame):
-        pricing_class = self.context.config.get('pricing_function_class', StaticPricer)
+    def transform(self, elasticity_df: pd.DataFrame):
+        pricing_class = self.context.config.get('pricing_function_class', ElasticityBasedPricer)
         pricing_params = self.context.config.get('pricing_function_params', {})
 
         pricer = pricing_class(**pricing_params)
-        pricer.fit(pricing_data)
+        pricer.fit(elasticity_df)
 
         return pricer
 

experiments/procesing/steps/session.py

@@ -1,261 +1,114 @@
 """
-Session feature extraction for ML training pipeline.
+Session feature extraction for S_t component of state space.
+Computes behavioral signals from interaction data already in pipeline.
 """
 import pandas as pd
 import numpy as np
-import re
-from typing import Dict, Any
+from typing import Optional, Dict, Any
+from collections import Counter
 from procesing.steps.base import BaseContextStep
 
-EVENT_CATS = {
-    'page_view': ['page_view'],
-    'item_view': ['view_item_page', 'learn_more_about_item'],
-    'cart_add': ['add_item_to_cart'],
-    'purchase': ['purchase', 'checkout_complete'],
-    'hover': ['hover_over_title', 'hover_over_paragraph', 'hover_over_link', 'hover_over_button'],
-    # 'filter': ['filter', 'search', 'apply_filter'],
-}
-HEADLESS_RE = re.compile(r'HeadlessChrome|Headless|PhantomJS', re.I)
-AUTOMATION_RE = re.compile(r'Selenium|Playwright|Puppeteer|WebDriver|chromedriver|geckodriver', re.I)
-BROWSER_PATTERNS = [('Chrome', r'Chrome/[\d.]+'), ('Firefox', r'Firefox/[\d.]+'),
-                    ('Safari', r'Safari/[\d.]+'), ('Edge', r'Edg/[\d.]+')]
-
-
-def _get_browser(s: str) -> str:
-    if pd.isna(s): return 'Unknown'
-    for name, pat in BROWSER_PATTERNS:
-        if re.search(pat, s): return name
-    return 'Other'
-
-
-class TemporalFeatureStep(BaseContextStep):
-    """Vectorized time-based features: durations, velocities, gaps."""
-
-    def __init__(self, context, timeout_sec: float = 900, velocity_window: str = '5min'):
-        super().__init__(context)
-        self.timeout_sec = timeout_sec
-        self.velocity_window = velocity_window
-
-    def transform(self, X: pd.DataFrame) -> pd.DataFrame:
-        df = X.copy()
-        if df.empty or 'ts' not in df.columns:
-            return pd.DataFrame(columns=pd.Series(['sessionId']))
-
-        df['ts_dt'] = pd.to_datetime(df['ts'])
-        df = df.sort_values(['sessionId', 'ts_dt'])
-        df['time_diff'] = df.groupby('sessionId')['ts_dt'].diff().dt.total_seconds()
-        df['active_diff'] = df['time_diff'].where(df['time_diff'] <= self.timeout_sec, 0)
-
-        agg = df.groupby('sessionId').agg(
-            session_duration_sec=('active_diff', 'sum'),
-            total_interactions=('sessionId', 'count'),
-            avg_time_between_events=('time_diff', 'mean'),
-            std_time_between_events=('time_diff', 'std'),
-            min_time_between_events=('time_diff', 'min'),
-            session_start_hour=('ts_dt', lambda x: x.min().hour),
-        ).reset_index()
-        agg['std_time_between_events'] = agg['std_time_between_events'].fillna(0)
-        agg['interaction_velocity'] = np.where(
-            agg['session_duration_sec'] > 0,
-            (agg['total_interactions'] / agg['session_duration_sec']) * 60, 0)
-
-        vel = df.set_index('ts_dt').groupby('sessionId').resample(self.velocity_window, include_groups=False).size()
-        max_velocity = vel.groupby('sessionId').max().rename('max_velocity_5min')
-        agg = agg.merge(max_velocity, on='sessionId', how='left')
-        agg['max_velocity_5min'] = agg['max_velocity_5min'].fillna(0)
-        return agg
-
-
-class BehavioralFeatureStep(BaseContextStep):
-    """Vectorized event counts and ratios per session."""
-
-    def transform(self, X: pd.DataFrame) -> pd.DataFrame:
-        df = X.copy()
-        if df.empty or 'eventName' not in df.columns:
-            return pd.DataFrame(columns=pd.Series(['sessionId']))
-
-        for cat, events in EVENT_CATS.items():
-            df[f'is_{cat}'] = df['eventName'].isin(events)
-        df['is_hover'] = df['is_hover'] | df['eventName'].str.startswith('hover_over_')
-
-        agg = df.groupby('sessionId').agg(
-            total_events=('eventName', 'count'), unique_pages=('page', 'nunique'),
-            page_views=('is_page_view', 'sum'), item_views=('is_item_view', 'sum'),
-            cart_adds=('is_cart_add', 'sum'), purchases=('is_purchase', 'sum'),
-            hover_events=('is_hover', 'sum'),
-            # filter_events=('is_filter', 'sum'),
-        ).reset_index()
-        agg['cart_to_view_ratio'] = np.where(agg['item_views'] > 0, agg['cart_adds'] / agg['item_views'], 0)
-        agg['conversion_rate'] = np.where(agg['item_views'] > 0, agg['purchases'] / agg['item_views'], 0)
-        agg['hover_intensity'] = np.where(agg['total_events'] > 0, agg['hover_events'] / agg['total_events'], 0)
-        return agg
-
-
-class ProductFeatureStep(BaseContextStep):
-    """Vectorized product interaction features: diversity, depth, price sensitivity."""
-
-    def transform(self, X: pd.DataFrame) -> pd.DataFrame:
-        df = X.copy()
-        if df.empty:
-            return pd.DataFrame(columns=pd.Series(['sessionId']))
-        price_col = next((c for c in ['metadata_base_price', 'metadata_price', 'base_price'] if c in df.columns), None)
-        df['price_seen'] = pd.to_numeric(df[price_col], errors='coerce') if price_col else np.nan
-
-        prod_df = df[df['productId'].notna()]
-        if prod_df.empty:
-            return pd.DataFrame(columns=pd.Series(['sessionId', 'unique_products_viewed', 'product_view_depth', 'avg_price_seen', 'min_price_seen', 'max_price_seen', 'price_range']))
-
-        agg = prod_df.groupby('sessionId').agg(
-            unique_products_viewed=('productId', 'nunique'),
-            product_view_depth=('productId', lambda x: x.value_counts().iloc[0] if len(x) > 0 else 0),
-            avg_price_seen=('price_seen', 'mean'), min_price_seen=('price_seen', 'min'),
-            max_price_seen=('price_seen', 'max'),
-        ).reset_index()
-        agg['price_range'] = (agg['max_price_seen'] - agg['min_price_seen']).fillna(0)
-        return agg
-
-
-class UserAgentFeatureStep(BaseContextStep):
-    """Parse userAgent into bot-detection signals."""
-
-    def transform(self, X: pd.DataFrame) -> pd.DataFrame|pd.Series:
-        df = X.copy()
-        if df.empty or 'userAgent' not in df.columns:
-            return pd.DataFrame(columns=pd.Series(['sessionId']))
-
-        ua = df.groupby('sessionId')['userAgent'].first().reset_index()
-        ua['is_headless'] = ua['userAgent'].str.contains(HEADLESS_RE, na=False)
-        ua['is_automation'] = ua['userAgent'].str.contains(AUTOMATION_RE, na=False)
-        ua['browser_family'] = ua['userAgent'].apply(_get_browser)
-        return ua[['sessionId', 'is_headless', 'is_automation', 'browser_family']]
-
 
 class ExtractSessionFeaturesStep(BaseContextStep):
     """
-    Vectorized session feature extraction - replaces O(n^2) per-row loop.
-    Input: interactions_df
-    Output: session-level feature matrix
+    Extract session-level behavioral features from interaction logs.
+
+    Input: interactions_df (user-interactions from earlier pipeline step)
+    Output: session_features DataFrame [sessionId, feature_1, feature_2, ...]
+
+    Features computed:
+        - total_interactions: count of all events
+        - page_views, item_views, searches, cart_adds: event type counts
+        - hovers: hover event counts
+        - unique_products_viewed: distinct product IDs
+        - interaction_velocity: events per minute
+        - session_duration_sec: time span of session
+        - avg_time_between_events: mean inter-event time
+        - product_view_depth: max views for single product (attention signal)
     """
 
-    def transform(self, X: pd.DataFrame) -> pd.DataFrame:
-        if X.empty:
+    def transform(self, interactions_df: pd.DataFrame) -> pd.DataFrame:
+        if interactions_df.empty:
             return pd.DataFrame()
-        df = X.copy()
 
-        # run all feature steps and merge on sessionId
-        temporal = TemporalFeatureStep(self.context).transform(df)
-        behavioral = BehavioralFeatureStep(self.context).transform(df)
-        product = ProductFeatureStep(self.context).transform(df)
-        ua = UserAgentFeatureStep(self.context).transform(df)
+        # ensure timestamp column
+        if 'ts' in interactions_df.columns:
+            interactions_df = interactions_df.copy()
+            interactions_df['ts'] = pd.to_datetime(interactions_df['ts'])
 
-        result = temporal
-        for other in [behavioral, product, ua]:
-            if not other.empty and 'sessionId' in other.columns:
-                result = result.merge(other, on='sessionId', how='left')
+        # group by session and compute features
+        session_features = []
+        for session_id, session_df in interactions_df.groupby('sessionId'):
+            features = self._extract_features_for_session(session_id, session_df)
+            session_features.append(features)
 
-        # carry forward experimentId for label joining
-        if 'experimentId' in df.columns:
-            exp_map = df.groupby('sessionId')['experimentId'].first()
-            result = result.merge(exp_map, on='sessionId', how='left')
+        return pd.DataFrame(session_features)
 
-        return result
+    def _extract_features_for_session(self, session_id: str, session_df: pd.DataFrame) -> Dict[str, Any]:
+        """Compute features for single session."""
+        features = {'sessionId': session_id}
 
+        # basic counts
+        features['total_interactions'] = len(session_df)
 
-class JoinLabelsStep(BaseContextStep):
-    """
-    Join experiment labels to session features.
-    Input: (features_df, experiments_df) or features_df (fetches experiments)
-    Output: labeled feature matrix with is_agent column
-    """
+        event_counts = session_df['eventName'].value_counts().to_dict()
+        features['page_views'] = event_counts.get('page_view', 0) + event_counts.get('view_item_page', 0)
+        features['item_views'] = event_counts.get('view_item_page', 0)
+        features['searches'] = event_counts.get('search', 0)
+        features['cart_adds'] = event_counts.get('add_item_to_cart', 0)
 
-    def transform(self, X : tuple) -> pd.DataFrame:
-        data = X;
-        if isinstance(data, tuple):
-            features_df, experiments_df = data
+        # hover events
+        hover_events = ['hover_over_title', 'hover_over_paragraph', 'hover_over_link', 'hover_over_button']
+        features['hovers'] = sum(event_counts.get(ev, 0) for ev in hover_events)
+
+        # product-level signals
+        product_ids = session_df['productId'].dropna()
+        features['unique_products_viewed'] = product_ids.nunique()
+
+        if len(product_ids) > 0:
+            product_view_counts = Counter(product_ids)
+            features['product_view_depth'] = max(product_view_counts.values())
         else:
-            features_df = data
-            if 'experimentId' not in features_df.columns:
-                return features_df
-            exp_ids = features_df['experimentId'].dropna().unique().tolist()
-            experiments_df = self.context.provider.fetch_experiments(exp_ids) if exp_ids else pd.DataFrame()
+            features['product_view_depth'] = 0
 
-        if features_df.empty:
-            return features_df
-        if experiments_df.empty:
-            features_df['is_agent'] = np.nan
-            return features_df
+        # temporal features
+        if 'ts' in session_df.columns:
+            timestamps = session_df['ts'].sort_values()
+            features['session_duration_sec'] = (timestamps.max() - timestamps.min()).total_seconds()
 
-        exp = experiments_df.copy()
-        if 'id' in exp.columns:
-            exp = exp.rename(columns={'id': 'experimentId'})
-        if 'xp_human_only' in exp.columns:
-            exp['is_agent'] = ~exp['xp_human_only']
+            if features['session_duration_sec'] > 0:
+                features['interaction_velocity'] = (features['total_interactions'] / features['session_duration_sec']) * 60
+            else:
+                features['interaction_velocity'] = 0.0
 
-        cols = ['experimentId'] + [c for c in ['is_agent', 'xp_human_only', 'xp_market_mode'] if c in exp.columns]
-        return features_df.merge(exp[cols].drop_duplicates(), on='experimentId', how='left')
+            # inter-event timing
+            if len(timestamps) > 1:
+                time_diffs = timestamps.diff().dropna().dt.total_seconds()
+                features['avg_time_between_events'] = time_diffs.mean()
+                features['std_time_between_events'] = time_diffs.std()
+            else:
+                features['avg_time_between_events'] = 0.0
+                features['std_time_between_events'] = 0.0
+        else:
+            features['session_duration_sec'] = 0.0
+            features['interaction_velocity'] = 0.0
+            features['avg_time_between_events'] = 0.0
+            features['std_time_between_events'] = 0.0
+
+        # cart/conversion signals
+        features['cart_to_view_ratio'] = features['cart_adds'] / features['item_views'] if features['item_views'] > 0 else 0.0
+
+        return features
 
 
-class ValidateDataStep(BaseContextStep):
+class FilterSessionInteractionsStep(BaseContextStep):
     """
-    Data quality checks before training.
-    Input: df
-    Output: df (unchanged, but logs validation report to context)
+    Filter interactions DataFrame to specific session.
+
+    Input: (interactions_df, session_id)
+    Output: interactions_df filtered to session_id
     """
-    REQUIRED = ['sessionId', 'eventName', 'ts']
 
-    def transform(self, X: pd.DataFrame) -> pd.DataFrame:
-        df = X.copy()
-        report = {'status': 'valid', 'rows': len(df), 'sessions': 0}
-        if df.empty:
-            report['status'] = 'empty'
-            self.context.cache('validation_report', report)
-            return df
-
-        missing = [c for c in self.REQUIRED if c not in df.columns]
-        if missing:
-            report['status'] = 'invalid'
-            report['missing_cols'] = missing
-
-        report['sessions'] = df['sessionId'].nunique() if 'sessionId' in df.columns else 0
-        report['null_sessions'] = int(df['sessionId'].isna().sum()) if 'sessionId' in df.columns else 0
-        if 'experimentId' in df.columns:
-            report['null_experiments'] = int(df['experimentId'].isna().sum())
-
-        self.context.cache('validation_report', report)
-        return df
-
-
-# legacy compat - kept for backwards compatibility with existing code
-def _extract_features_for_session(session_df: pd.DataFrame, session_timeout_sec: float = 900) -> Dict[str, Any]:
-    """Single-session feature extraction (legacy interface)."""
-    defaults = {k: 0 for k in ['total_interactions', 'page_views', 'item_views', 'searches',
-                               'cart_adds', 'hovers', 'unique_products_viewed', 'product_view_depth',
-                               'session_duration_sec', 'interaction_velocity',
-                               'avg_time_between_events', 'std_time_between_events', 'cart_to_view_ratio']}
-    if session_df.empty:
-        return defaults
-
-    session_df = session_df.copy()
-    if 'sessionId' not in session_df.columns:
-        session_df['sessionId'] = 'tmp'
-
-    # use a dummy context for the steps
-    class DummyCtx: config = {} # should maybe inherit but whatever
-    ctx = DummyCtx()
-
-    t = TemporalFeatureStep(ctx, timeout_sec=session_timeout_sec).transform(session_df)
-    b = BehavioralFeatureStep(ctx).transform(session_df)
-    p = ProductFeatureStep(ctx).transform(session_df)
-
-    result = {}
-    for df in [t, b, p]:
-        if not df.empty:
-            for col in df.columns:
-                if col != 'sessionId':
-                    result[col] = df[col].iloc[0] if len(df) > 0 else 0
-
-    remap = {'hover_events': 'hovers', 'filter_events': 'searches', 'unique_pages': 'unique_pages_visited'}
-    for old, new in remap.items():
-        if old in result:
-            result[new] = result.pop(old)
-    return result
+    def transform(self, data: tuple) -> pd.DataFrame:
+        interactions_df, session_id = data
+        return interactions_df[interactions_df['sessionId'] == session_id].copy()

experiments/procesing/tests/conftest.py

@@ -144,7 +144,7 @@ def mock_price_logs_raw_kafka():
                     'price': 162.47,
                     'sessionId': 'd423ce8a-77aa-4c9a-94d4-d1adddcc3472',
                     'experimentId': '53aefd07-f66a-4d7f-ba8b-7ea1fc562d35',
-                    'storeMode': 'hotel',
+                    'storeMode': 'shop',
                     'ts': '2025-11-25T21:05:57.967Z'
                 }
             }
@@ -157,7 +157,7 @@ def mock_price_logs_raw_kafka():
                     'price': 743.49,
                     'sessionId': 'd423ce8a-77aa-4c9a-94d4-d1adddcc3472',
                     'experimentId': '53aefd07-f66a-4d7f-ba8b-7ea1fc562d35',
-                    'storeMode': 'hotel',
+                    'storeMode': 'shop',
                     'ts': '2025-11-25T21:05:57.993Z'
                 }
             }
@@ -170,7 +170,7 @@ def mock_price_logs_raw_kafka():
                     'price': 163.87,
                     'sessionId': 'd423ce8a-77aa-4c9a-94d4-d1adddcc3472',
                     'experimentId': '53aefd07-f66a-4d7f-ba8b-7ea1fc562d35',
-                    'storeMode': 'hotel',
+                    'storeMode': 'shop',
                     'ts': '2025-11-25T21:05:58.009Z'
                 }
             }
@@ -183,7 +183,7 @@ def mock_price_logs_raw_kafka():
                     'price': 397.46,
                     'sessionId': 'd423ce8a-77aa-4c9a-94d4-d1adddcc3472',
                     'experimentId': '53aefd07-f66a-4d7f-ba8b-7ea1fc562d35',
-                    'storeMode': 'hotel',
+                    'storeMode': 'shop',
                     'ts': '2025-11-25T21:05:58.049Z'
                 }
             }
@@ -196,7 +196,7 @@ def mock_price_logs_raw_kafka():
                     'price': 401.66,
                     'sessionId': 'd423ce8a-77aa-4c9a-94d4-d1adddcc3472',
                     'experimentId': '53aefd07-f66a-4d7f-ba8b-7ea1fc562d35',
-                    'storeMode': 'hotel',
+                    'storeMode': 'shop',
                     'ts': '2025-11-25T21:06:08.864Z'
                 }
             }
@@ -222,7 +222,7 @@ def mock_experiments():
         'created_at': pd.to_datetime(['2025-11-25T20:00:00Z', '2025-11-26T10:00:00Z']),
         'subject_name': ['Session A', 'Session B'],
         'xp_human_only': [True, False],
-        'xp_market_mode': ['hotel', 'airline'],
+        'xp_market_mode': ['hotel', 'shop'],
         'xp_task_id': [None, None]
     })
 
@@ -269,13 +269,3 @@ def empty_context(empty_provider):
         store_mode='hotel',
         window_size='30s'
     )
-
-
-@pytest.fixture
-def session_interactions(mock_interactions):
-    """Enriched interaction data for session feature extraction tests"""
-    df = mock_interactions.copy()
-    df['userAgent'] = ['Mozilla/5.0 Chrome/120', 'Mozilla/5.0 Chrome/120',
-                       'HeadlessChrome/120', 'HeadlessChrome/120', 'HeadlessChrome/120']
-    df['metadata_base_price'] = [None, None, 150.0, 150.0, 200.0]
-    return df

experiments/procesing/tests/test_elasticity.py

@@ -0,0 +1,353 @@
+import pytest
+import pandas as pd
+import numpy as np
+from procesing.steps import (
+    AggregatePriceLogsStep,
+    ComputeElasticityStep
+)
+
+
+def test_aggregate_price_logs_basic(pipeline_context):
+    """Test basic price aggregation into time windows"""
+    step = AggregatePriceLogsStep(pipeline_context)
+
+    # Create price logs with known window structure
+    df = pd.DataFrame({
+        'ts': pd.date_range(start='2023-01-01 10:00:00', periods=100, freq='10s'),
+        'productId': np.tile([
+            'd018efc1-25e9-4284-b276-80386e048b25',
+            '51266ddb-5b07-47b7-89ee-5b5cae94bb11',
+            '2cd7f756-fc65-4ba0-ab01-74521c1fff43'
+        ], 34)[:100],
+        'price': np.random.uniform(100, 200, 100)
+    })
+
+    result = step.transform(df)
+    assert isinstance(result, list)
+    assert len(result) > 0
+    # each chunk should have window metadata and price vector
+    for chunk in result:
+        assert 'window_start' in chunk
+        assert 'window_end' in chunk
+        assert 'price_vector' in chunk
+        assert isinstance(chunk['price_vector'], pd.DataFrame)
+        assert 'productId' in chunk['price_vector'].columns
+        assert 'price' in chunk['price_vector'].columns
+
+
+def test_aggregate_price_logs_handles_gaps(pipeline_context):
+    """Test that price aggregation forward-fills missing windows"""
+    step = AggregatePriceLogsStep(pipeline_context)
+
+    # create sparse data with gaps
+    df = pd.DataFrame({
+        'ts': pd.to_datetime([
+            '2023-01-01 10:00:00',
+            '2023-01-01 10:00:05',
+            '2023-01-01 10:02:00',  # gap of ~2 mins
+            '2023-01-01 10:02:30'
+        ]),
+        'productId': [
+            'd018efc1-25e9-4284-b276-80386e048b25',
+            'd018efc1-25e9-4284-b276-80386e048b25',
+            '51266ddb-5b07-47b7-89ee-5b5cae94bb11',
+            '51266ddb-5b07-47b7-89ee-5b5cae94bb11'
+        ],
+        'price': [100, 102, 150, 153]
+    })
+
+    result = step.transform(df)
+    assert isinstance(result, list)
+    # should have multiple windows despite gaps
+    assert len(result) >= 2
+
+
+def test_compute_elasticity_with_known_relationship(pipeline_context):
+    """Test elasticity computation with known price-demand relationship"""
+    step = ComputeElasticityStep(pipeline_context)
+
+    # simulate elastic demand: when price ↑10%, demand ↓15% (elasticity ~ -1.5)
+    base_price = 100
+    base_demand = 50
+
+    demand_chunks = [
+        {
+            'window_start': pd.Timestamp('2023-01-01 10:00:00'),
+            'window_end': pd.Timestamp('2023-01-01 10:00:30'),
+            'demand_vector': pd.DataFrame({
+                'productId': ['d018efc1-25e9-4284-b276-80386e048b25'],
+                'demand_score': [base_demand]
+            })
+        },
+        {
+            'window_start': pd.Timestamp('2023-01-01 10:00:30'),
+            'window_end': pd.Timestamp('2023-01-01 10:01:00'),
+            'demand_vector': pd.DataFrame({
+                'productId': ['d018efc1-25e9-4284-b276-80386e048b25'],
+                'demand_score': [base_demand * 0.85]  # 15% decrease
+            })
+        },
+        {
+            'window_start': pd.Timestamp('2023-01-01 10:01:00'),
+            'window_end': pd.Timestamp('2023-01-01 10:01:30'),
+            'demand_vector': pd.DataFrame({
+                'productId': ['d018efc1-25e9-4284-b276-80386e048b25'],
+                'demand_score': [base_demand * 0.70]  # further decrease
+            })
+        }
+    ]
+
+    price_chunks = [
+        {
+            'window_start': pd.Timestamp('2023-01-01 10:00:00'),
+            'window_end': pd.Timestamp('2023-01-01 10:00:30'),
+            'price_vector': pd.DataFrame({
+                'productId': ['d018efc1-25e9-4284-b276-80386e048b25'],
+                'price': [base_price]
+            })
+        },
+        {
+            'window_start': pd.Timestamp('2023-01-01 10:00:30'),
+            'window_end': pd.Timestamp('2023-01-01 10:01:00'),
+            'price_vector': pd.DataFrame({
+                'productId': ['d018efc1-25e9-4284-b276-80386e048b25'],
+                'price': [base_price * 1.10]  # 10% increase
+            })
+        },
+        {
+            'window_start': pd.Timestamp('2023-01-01 10:01:00'),
+            'window_end': pd.Timestamp('2023-01-01 10:01:30'),
+            'price_vector': pd.DataFrame({
+                'productId': ['d018efc1-25e9-4284-b276-80386e048b25'],
+                'price': [base_price * 1.20]  # 20% increase
+            })
+        }
+    ]
+
+    result = step.transform((demand_chunks, price_chunks))
+    assert isinstance(result, pd.DataFrame)
+    assert not result.empty
+    assert 'productId' in result.columns
+    assert 'elasticity' in result.columns
+    assert 'n_obs' in result.columns
+
+    # check elasticity is negative (normal good)
+    product_elast = result[result['productId'] == 'd018efc1-25e9-4284-b276-80386e048b25']
+    assert len(product_elast) == 1
+    assert product_elast.iloc[0]['elasticity'] < 0
+    # should be roughly elastic (< -1)
+    assert product_elast.iloc[0]['n_obs'] == 3
+
+
+def test_compute_elasticity_inelastic_product(pipeline_context):
+    """Test with inelastic demand: price changes, demand barely moves"""
+    step = ComputeElasticityStep(pipeline_context)
+
+    base_price = 150
+    base_demand = 40
+
+    demand_chunks = [
+        {
+            'window_start': pd.Timestamp('2023-01-01 10:00:00'),
+            'window_end': pd.Timestamp('2023-01-01 10:00:30'),
+            'demand_vector': pd.DataFrame({
+                'productId': ['51266ddb-5b07-47b7-89ee-5b5cae94bb11'],
+                'demand_score': [base_demand]
+            })
+        },
+        {
+            'window_start': pd.Timestamp('2023-01-01 10:00:30'),
+            'window_end': pd.Timestamp('2023-01-01 10:01:00'),
+            'demand_vector': pd.DataFrame({
+                'productId': ['51266ddb-5b07-47b7-89ee-5b5cae94bb11'],
+                'demand_score': [base_demand * 0.98]  # tiny 2% decrease
+            })
+        }
+    ]
+
+    price_chunks = [
+        {
+            'window_start': pd.Timestamp('2023-01-01 10:00:00'),
+            'window_end': pd.Timestamp('2023-01-01 10:00:30'),
+            'price_vector': pd.DataFrame({
+                'productId': ['51266ddb-5b07-47b7-89ee-5b5cae94bb11'],
+                'price': [base_price]
+            })
+        },
+        {
+            'window_start': pd.Timestamp('2023-01-01 10:00:30'),
+            'window_end': pd.Timestamp('2023-01-01 10:01:00'),
+            'price_vector': pd.DataFrame({
+                'productId': ['51266ddb-5b07-47b7-89ee-5b5cae94bb11'],
+                'price': [base_price * 1.20]  # 20% increase
+            })
+        }
+    ]
+
+    result = step.transform((demand_chunks, price_chunks))
+    product_elast = result[result['productId'] == '51266ddb-5b07-47b7-89ee-5b5cae94bb11']
+    assert len(product_elast) == 1
+    # inelastic: elasticity between 0 and -1
+    assert -1 < product_elast.iloc[0]['elasticity'] < 0
+
+
+def test_compute_elasticity_multiple_products(pipeline_context):
+    """Test elasticity computation across multiple products simultaneously"""
+    step = ComputeElasticityStep(pipeline_context)
+
+    products = [
+        'd018efc1-25e9-4284-b276-80386e048b25',
+        '51266ddb-5b07-47b7-89ee-5b5cae94bb11',
+        '2cd7f756-fc65-4ba0-ab01-74521c1fff43'
+    ]
+
+    # create 5 time windows with all 3 products
+    demand_chunks = []
+    price_chunks = []
+
+    for i in range(5):
+        ts = pd.Timestamp('2023-01-01 10:00:00') + pd.Timedelta(f'{i*30}s')
+
+        demand_chunks.append({
+            'window_start': ts,
+            'window_end': ts + pd.Timedelta('30s'),
+            'demand_vector': pd.DataFrame({
+                'productId': products,
+                'demand_score': [
+                    50 * (0.9 ** i),  # elastic: decreases as price rises
+                    40 * (0.98 ** i), # inelastic: barely changes
+                    30 * (0.85 ** i)  # very elastic
+                ]
+            })
+        })
+
+        price_chunks.append({
+            'window_start': ts,
+            'window_end': ts + pd.Timedelta('30s'),
+            'price_vector': pd.DataFrame({
+                'productId': products,
+                'price': [
+                    100 * (1.05 ** i),
+                    150 * (1.10 ** i),
+                    120 * (1.08 ** i)
+                ]
+            })
+        })
+
+    result = step.transform((demand_chunks, price_chunks))
+    assert isinstance(result, pd.DataFrame)
+    assert len(result) == 3  # all products should have elasticity
+    assert set(result['productId']) == set(products)
+    assert all(result['n_obs'] == 5)
+    assert all(result['elasticity'] < 0)  # all normal goods
+
+
+def test_compute_elasticity_insufficient_data(pipeline_context):
+    """Test behavior with insufficient observations"""
+    step = ComputeElasticityStep(pipeline_context)
+
+    # only 1 observation
+    demand_chunks = [{
+        'window_start': pd.Timestamp('2023-01-01 10:00:00'),
+        'window_end': pd.Timestamp('2023-01-01 10:00:30'),
+        'demand_vector': pd.DataFrame({
+            'productId': ['d018efc1-25e9-4284-b276-80386e048b25'],
+            'demand_score': [50]
+        })
+    }]
+
+    price_chunks = [{
+        'window_start': pd.Timestamp('2023-01-01 10:00:00'),
+        'window_end': pd.Timestamp('2023-01-01 10:00:30'),
+        'price_vector': pd.DataFrame({
+            'productId': ['d018efc1-25e9-4284-b276-80386e048b25'],
+            'price': [100]
+        })
+    }]
+
+    result = step.transform((demand_chunks, price_chunks))
+    # should still return result but with low n_obs
+    product_elast = result[result['productId'] == 'd018efc1-25e9-4284-b276-80386e048b25']
+    assert len(product_elast) == 1
+    assert product_elast.iloc[0]['n_obs'] == 1
+    assert product_elast.iloc[0]['elasticity'] == 0.0  # not enough data
+
+
+def test_compute_elasticity_misaligned_chunks(pipeline_context):
+    """Test with non-overlapping demand and price windows"""
+    step = ComputeElasticityStep(pipeline_context)
+
+    demand_chunks = [{
+        'window_start': pd.Timestamp('2023-01-01 10:00:00'),
+        'window_end': pd.Timestamp('2023-01-01 10:00:30'),
+        'demand_vector': pd.DataFrame({
+            'productId': ['d018efc1-25e9-4284-b276-80386e048b25'],
+            'demand_score': [50]
+        })
+    }]
+
+    price_chunks = [{
+        'window_start': pd.Timestamp('2023-01-01 11:00:00'),  # different time
+        'window_end': pd.Timestamp('2023-01-01 11:00:30'),
+        'price_vector': pd.DataFrame({
+            'productId': ['d018efc1-25e9-4284-b276-80386e048b25'],
+            'price': [100]
+        })
+    }]
+
+    result = step.transform((demand_chunks, price_chunks))
+    # should handle gracefully with no aligned data
+    assert isinstance(result, pd.DataFrame)
+    assert all(result['n_obs'] == 0)
+
+
+def test_elasticity_arc_method(pipeline_context):
+    """Test arc elasticity computation method"""
+    # configure context for arc method
+    pipeline_context.config['elasticity_method'] = 'arc'
+    step = ComputeElasticityStep(pipeline_context)
+
+    demand_chunks = [
+        {
+            'window_start': pd.Timestamp('2023-01-01 10:00:00'),
+            'window_end': pd.Timestamp('2023-01-01 10:00:30'),
+            'demand_vector': pd.DataFrame({
+                'productId': ['d018efc1-25e9-4284-b276-80386e048b25'],
+                'demand_score': [100]
+            })
+        },
+        {
+            'window_start': pd.Timestamp('2023-01-01 10:00:30'),
+            'window_end': pd.Timestamp('2023-01-01 10:01:00'),
+            'demand_vector': pd.DataFrame({
+                'productId': ['d018efc1-25e9-4284-b276-80386e048b25'],
+                'demand_score': [80]
+            })
+        }
+    ]
+
+    price_chunks = [
+        {
+            'window_start': pd.Timestamp('2023-01-01 10:00:00'),
+            'window_end': pd.Timestamp('2023-01-01 10:00:30'),
+            'price_vector': pd.DataFrame({
+                'productId': ['d018efc1-25e9-4284-b276-80386e048b25'],
+                'price': [100]
+            })
+        },
+        {
+            'window_start': pd.Timestamp('2023-01-01 10:00:30'),
+            'window_end': pd.Timestamp('2023-01-01 10:01:00'),
+            'price_vector': pd.DataFrame({
+                'productId': ['d018efc1-25e9-4284-b276-80386e048b25'],
+                'price': [110]
+            })
+        }
+    ]
+
+    result = step.transform((demand_chunks, price_chunks))
+    product_elast = result[result['productId'] == 'd018efc1-25e9-4284-b276-80386e048b25']
+    assert len(product_elast) == 1
+    assert product_elast.iloc[0]['elasticity'] < 0
+    # reset config
+    pipeline_context.config['elasticity_method'] = 'point'

lib/model_registry.py

@@ -26,7 +26,6 @@ class ModelRegistry:
         self.metadata_prefix = "model:meta:"
         self.data_prefix = "model:data:"
         self.elasticity_prefix = "elasticity:"
-        self.prices_prefix = "prices:"
 
     def publish_elasticity(self,
                           elasticity_df: pd.DataFrame,
@@ -131,46 +130,6 @@ class ModelRegistry:
 
         return models
 
-    def publish_prices(self,
-                      prices_df: pd.DataFrame,
-                      model_name: str = 'latest',
-                      metadata: Optional[Dict[str, Any]] = None):
-        """Store predicted prices in registry.
-
-        Args:
-            prices_df: df with [productId, predicted_price, ...]
-            model_name: identifier for this price snapshot
-            metadata: additional info
-        """
-        key = f"{self.prices_prefix}{model_name}"
-        data_json = prices_df.to_json(orient='records')
-
-        self.redis_client.set(key, data_json)
-
-        meta = metadata or {}
-        meta.update({
-            'n_products': len(prices_df),
-            'model_type': 'predicted_prices'
-        })
-
-        meta_key = f"{self.metadata_prefix}prices_{model_name}"
-        self.redis_client.set(meta_key, json.dumps(meta))
-
-        log.info(f"Published prices '{model_name}' for {len(prices_df)} products")
-
-    def get_prices(self, model_name: str = 'latest') -> Optional[pd.DataFrame]:
-        """Retrieve predicted prices from registry."""
-        key = f"{self.prices_prefix}{model_name}"
-        data_json = self.redis_client.get(key)
-
-        if data_json is None:
-            return None
-
-        if isinstance(data_json, bytes):
-            data_json = data_json.decode('utf-8')
-
-        return pd.read_json(data_json, orient='records')
-
     def health_check(self) -> bool:
         """Check if Redis connection is alive."""
         try:

requirements.txt

@@ -11,4 +11,3 @@ pytest-asyncio
 uv
 scikit-learn
 supabase
-pymc

web/package-lock.json

@@ -10,7 +10,7 @@
       "dependencies": {
         "@supabase/ssr": "^0.7.0",
         "@supabase/supabase-js": "^2.81.1",
-        "next": "^16.0.0",
+        "next": "16.0.0",
         "react": "19.2.0",
         "react-dom": "19.2.0",
         "zod": "^4.1.12"
@@ -526,15 +526,15 @@
       }
     },
     "node_modules/@next/env": {
-      "version": "16.0.7",
-      "resolved": "https://registry.npmjs.org/@next/env/-/env-16.0.7.tgz",
-      "integrity": "sha512-gpaNgUh5nftFKRkRQGnVi5dpcYSKGcZZkQffZ172OrG/XkrnS7UBTQ648YY+8ME92cC4IojpI2LqTC8sTDhAaw==",
+      "version": "16.0.0",
+      "resolved": "https://registry.npmjs.org/@next/env/-/env-16.0.0.tgz",
+      "integrity": "sha512-s5j2iFGp38QsG1LWRQaE2iUY3h1jc014/melHFfLdrsMJPqxqDQwWNwyQTcNoUSGZlCVZuM7t7JDMmSyRilsnA==",
       "license": "MIT"
     },
     "node_modules/@next/swc-darwin-arm64": {
-      "version": "16.0.7",
-      "resolved": "https://registry.npmjs.org/@next/swc-darwin-arm64/-/swc-darwin-arm64-16.0.7.tgz",
-      "integrity": "sha512-LlDtCYOEj/rfSnEn/Idi+j1QKHxY9BJFmxx7108A6D8K0SB+bNgfYQATPk/4LqOl4C0Wo3LACg2ie6s7xqMpJg==",
+      "version": "16.0.0",
+      "resolved": "https://registry.npmjs.org/@next/swc-darwin-arm64/-/swc-darwin-arm64-16.0.0.tgz",
+      "integrity": "sha512-/CntqDCnk5w2qIwMiF0a9r6+9qunZzFmU0cBX4T82LOflE72zzH6gnOjCwUXYKOBlQi8OpP/rMj8cBIr18x4TA==",
       "cpu": [
         "arm64"
       ],
@@ -548,9 +548,9 @@
       }
     },
     "node_modules/@next/swc-darwin-x64": {
-      "version": "16.0.7",
-      "resolved": "https://registry.npmjs.org/@next/swc-darwin-x64/-/swc-darwin-x64-16.0.7.tgz",
-      "integrity": "sha512-rtZ7BhnVvO1ICf3QzfW9H3aPz7GhBrnSIMZyr4Qy6boXF0b5E3QLs+cvJmg3PsTCG2M1PBoC+DANUi4wCOKXpA==",
+      "version": "16.0.0",
+      "resolved": "https://registry.npmjs.org/@next/swc-darwin-x64/-/swc-darwin-x64-16.0.0.tgz",
+      "integrity": "sha512-hB4GZnJGKa8m4efvTGNyii6qs76vTNl+3dKHTCAUaksN6KjYy4iEO3Q5ira405NW2PKb3EcqWiRaL9DrYJfMHg==",
       "cpu": [
         "x64"
       ],
@@ -564,9 +564,9 @@
       }
     },
     "node_modules/@next/swc-linux-arm64-gnu": {
-      "version": "16.0.7",
-      "resolved": "https://registry.npmjs.org/@next/swc-linux-arm64-gnu/-/swc-linux-arm64-gnu-16.0.7.tgz",
-      "integrity": "sha512-mloD5WcPIeIeeZqAIP5c2kdaTa6StwP4/2EGy1mUw8HiexSHGK/jcM7lFuS3u3i2zn+xH9+wXJs6njO7VrAqww==",
+      "version": "16.0.0",
+      "resolved": "https://registry.npmjs.org/@next/swc-linux-arm64-gnu/-/swc-linux-arm64-gnu-16.0.0.tgz",
+      "integrity": "sha512-E2IHMdE+C1k+nUgndM13/BY/iJY9KGCphCftMh7SXWcaQqExq/pJU/1Hgn8n/tFwSoLoYC/yUghOv97tAsIxqg==",
       "cpu": [
         "arm64"
       ],
@@ -580,9 +580,9 @@
       }
     },
     "node_modules/@next/swc-linux-arm64-musl": {
-      "version": "16.0.7",
-      "resolved": "https://registry.npmjs.org/@next/swc-linux-arm64-musl/-/swc-linux-arm64-musl-16.0.7.tgz",
-      "integrity": "sha512-+ksWNrZrthisXuo9gd1XnjHRowCbMtl/YgMpbRvFeDEqEBd523YHPWpBuDjomod88U8Xliw5DHhekBC3EOOd9g==",
+      "version": "16.0.0",
+      "resolved": "https://registry.npmjs.org/@next/swc-linux-arm64-musl/-/swc-linux-arm64-musl-16.0.0.tgz",
+      "integrity": "sha512-xzgl7c7BVk4+7PDWldU+On2nlwnGgFqJ1siWp3/8S0KBBLCjonB6zwJYPtl4MUY7YZJrzzumdUpUoquu5zk8vg==",
       "cpu": [
         "arm64"
       ],
@@ -596,9 +596,9 @@
       }
     },
     "node_modules/@next/swc-linux-x64-gnu": {
-      "version": "16.0.7",
-      "resolved": "https://registry.npmjs.org/@next/swc-linux-x64-gnu/-/swc-linux-x64-gnu-16.0.7.tgz",
-      "integrity": "sha512-4WtJU5cRDxpEE44Ana2Xro1284hnyVpBb62lIpU5k85D8xXxatT+rXxBgPkc7C1XwkZMWpK5rXLXTh9PFipWsA==",
+      "version": "16.0.0",
+      "resolved": "https://registry.npmjs.org/@next/swc-linux-x64-gnu/-/swc-linux-x64-gnu-16.0.0.tgz",
+      "integrity": "sha512-sdyOg4cbiCw7YUr0F/7ya42oiVBXLD21EYkSwN+PhE4csJH4MSXUsYyslliiiBwkM+KsuQH/y9wuxVz6s7Nstg==",
       "cpu": [
         "x64"
       ],
@@ -612,9 +612,9 @@
       }
     },
     "node_modules/@next/swc-linux-x64-musl": {
-      "version": "16.0.7",
-      "resolved": "https://registry.npmjs.org/@next/swc-linux-x64-musl/-/swc-linux-x64-musl-16.0.7.tgz",
-      "integrity": "sha512-HYlhqIP6kBPXalW2dbMTSuB4+8fe+j9juyxwfMwCe9kQPPeiyFn7NMjNfoFOfJ2eXkeQsoUGXg+O2SE3m4Qg2w==",
+      "version": "16.0.0",
+      "resolved": "https://registry.npmjs.org/@next/swc-linux-x64-musl/-/swc-linux-x64-musl-16.0.0.tgz",
+      "integrity": "sha512-IAXv3OBYqVaNOgyd3kxR4L3msuhmSy1bcchPHxDOjypG33i2yDWvGBwFD94OuuTjjTt/7cuIKtAmoOOml6kfbg==",
       "cpu": [
         "x64"
       ],
@@ -628,9 +628,9 @@
       }
     },
     "node_modules/@next/swc-win32-arm64-msvc": {
-      "version": "16.0.7",
-      "resolved": "https://registry.npmjs.org/@next/swc-win32-arm64-msvc/-/swc-win32-arm64-msvc-16.0.7.tgz",
-      "integrity": "sha512-EviG+43iOoBRZg9deGauXExjRphhuYmIOJ12b9sAPy0eQ6iwcPxfED2asb/s2/yiLYOdm37kPaiZu8uXSYPs0Q==",
+      "version": "16.0.0",
+      "resolved": "https://registry.npmjs.org/@next/swc-win32-arm64-msvc/-/swc-win32-arm64-msvc-16.0.0.tgz",
+      "integrity": "sha512-bmo3ncIJKUS9PWK1JD9pEVv0yuvp1KPuOsyJTHXTv8KDrEmgV/K+U0C75rl9rhIaODcS7JEb6/7eJhdwXI0XmA==",
       "cpu": [
         "arm64"
       ],
@@ -644,9 +644,9 @@
       }
     },
     "node_modules/@next/swc-win32-x64-msvc": {
-      "version": "16.0.7",
-      "resolved": "https://registry.npmjs.org/@next/swc-win32-x64-msvc/-/swc-win32-x64-msvc-16.0.7.tgz",
-      "integrity": "sha512-gniPjy55zp5Eg0896qSrf3yB1dw4F/3s8VK1ephdsZZ129j2n6e1WqCbE2YgcKhW9hPB9TVZENugquWJD5x0ug==",
+      "version": "16.0.0",
+      "resolved": "https://registry.npmjs.org/@next/swc-win32-x64-msvc/-/swc-win32-x64-msvc-16.0.0.tgz",
+      "integrity": "sha512-O1cJbT+lZp+cTjYyZGiDwsOjO3UHHzSqobkPNipdlnnuPb1swfcuY6r3p8dsKU4hAIEO4cO67ZCfVVH/M1ETXA==",
       "cpu": [
         "x64"
       ],
@@ -1447,12 +1447,12 @@
       }
     },
     "node_modules/next": {
-      "version": "16.0.7",
-      "resolved": "https://registry.npmjs.org/next/-/next-16.0.7.tgz",
-      "integrity": "sha512-3mBRJyPxT4LOxAJI6IsXeFtKfiJUbjCLgvXO02fV8Wy/lIhPvP94Fe7dGhUgHXcQy4sSuYwQNcOLhIfOm0rL0A==",
+      "version": "16.0.0",
+      "resolved": "https://registry.npmjs.org/next/-/next-16.0.0.tgz",
+      "integrity": "sha512-nYohiNdxGu4OmBzggxy9rczmjIGI+TpR5vbKTsE1HqYwNm1B+YSiugSrFguX6omMOKnDHAmBPY4+8TNJk0Idyg==",
       "license": "MIT",
       "dependencies": {
-        "@next/env": "16.0.7",
+        "@next/env": "16.0.0",
         "@swc/helpers": "0.5.15",
         "caniuse-lite": "^1.0.30001579",
         "postcss": "8.4.31",
@@ -1465,14 +1465,14 @@
         "node": ">=20.9.0"
       },
       "optionalDependencies": {
-        "@next/swc-darwin-arm64": "16.0.7",
-        "@next/swc-darwin-x64": "16.0.7",
-        "@next/swc-linux-arm64-gnu": "16.0.7",
-        "@next/swc-linux-arm64-musl": "16.0.7",
-        "@next/swc-linux-x64-gnu": "16.0.7",
-        "@next/swc-linux-x64-musl": "16.0.7",
-        "@next/swc-win32-arm64-msvc": "16.0.7",
-        "@next/swc-win32-x64-msvc": "16.0.7",
+        "@next/swc-darwin-arm64": "16.0.0",
+        "@next/swc-darwin-x64": "16.0.0",
+        "@next/swc-linux-arm64-gnu": "16.0.0",
+        "@next/swc-linux-arm64-musl": "16.0.0",
+        "@next/swc-linux-x64-gnu": "16.0.0",
+        "@next/swc-linux-x64-musl": "16.0.0",
+        "@next/swc-win32-arm64-msvc": "16.0.0",
+        "@next/swc-win32-x64-msvc": "16.0.0",
         "sharp": "^0.34.4"
       },
       "peerDependencies": {

web/package.json

@@ -10,7 +10,7 @@
   "dependencies": {
     "@supabase/ssr": "^0.7.0",
     "@supabase/supabase-js": "^2.81.1",
-    "next": "^16.0.0",
+    "next": "16.0.0",
     "react": "19.2.0",
     "react-dom": "19.2.0",
     "zod": "^4.1.12"

web/src/app/airline/checkout/page.tsx

@@ -1,11 +0,0 @@
-export default function AirlineCheckout() {
-  return (
-    <div className="min-h-screen flex items-center justify-center bg-gradient-to-br from-sky-50 to-blue-50">
-      <div className="text-center p-8">
-        <h1 className="text-4xl font-light text-gray-800 mb-4">
-          Thank you for flying with us
-        </h1>
-      </div>
-    </div>
-  );
-}

web/src/app/api/ingest/route.ts

@@ -7,7 +7,7 @@ export async function POST(req: NextRequest) {
   try {
     const body = await req.json();
 
-    const storeMode = process.env.NEXT_PUBLIC_STORE_MODE || process.env.STORE_MODE || 'hotel';
+    const storeMode = process.env.STORE_MODE || 'hotel';
     const userAgent = req.headers.get('user-agent') || undefined;
 
     const event: EventBase = {

web/src/app/api/pricing/route.ts

@@ -11,7 +11,7 @@ export async function GET(req: NextRequest) {
     const productId = searchParams.get('productId');
     const sessionId = searchParams.get('sessionId');
     const experimentId = searchParams.get('experimentId');
-    const storeMode = process.env.NEXT_PUBLIC_STORE_MODE || process.env.STORE_MODE || 'hotel';
+    const storeMode = process.env.NEXT_PUBLIC_STORE_MODE || 'shop';
 
     if (!productId) {
         return NextResponse.json(

web/src/app/cart/page.tsx

@@ -96,10 +96,7 @@ export default function CartPage() {
                                 <span className="text-3xl font-bold">${total.toFixed(2)}</span>
                             </div>
                             <button
-                                onClick={() => {
-                                    dispatchInteraction('checkout_start', undefined, { total, itemCount });
-                                    window.location.href = '/checkout';
-                                }}
+                                onClick={() => dispatchInteraction('checkout_start', undefined, { total, itemCount })}
                                 className="w-full py-3 bg-blue-600 hover:bg-blue-700 text-white rounded-lg font-medium transition-colors"
                             >
                                 Proceed to Checkout

web/src/app/hotel/checkout/page.tsx

@@ -1,11 +0,0 @@
-export default function HotelCheckout() {
-  return (
-    <div className="min-h-screen flex items-center justify-center bg-gradient-to-br from-blue-50 to-indigo-50">
-      <div className="text-center p-8">
-        <h1 className="text-4xl font-light text-gray-800 mb-4">
-          Thank you for staying with us
-        </h1>
-      </div>
-    </div>
-  );
-}

web/src/components/feats/airline/AirlineHero.tsx

@@ -2,20 +2,10 @@
 
 import { useState, FormEvent } from 'react';
 import { useRouter } from 'next/navigation';
-import { Button, Label, DateInput, Dropdown, DropdownCounter, SelectDropdown, SelectOption } from '@/components/ui';
+import { Button, Label, Input, DateInput, RadioGroup, Dropdown, DropdownCounter } from '@/components/ui';
 import { dateToDaysFromToday } from '@/lib/airline-utils';
 
-const CITIES: SelectOption[] = [
-  { value: 'JFK', label: 'New York (JFK)', sublabel: 'John F. Kennedy International' },
-  { value: 'LAX', label: 'Los Angeles (LAX)', sublabel: 'Los Angeles International' },
-  { value: 'ORD', label: 'Chicago (ORD)', sublabel: "O'Hare International" },
-  { value: 'MIA', label: 'Miami (MIA)', sublabel: 'Miami International' },
-  { value: 'SFO', label: 'San Francisco (SFO)', sublabel: 'San Francisco International' },
-  { value: 'SEA', label: 'Seattle (SEA)', sublabel: 'Seattle-Tacoma International' },
-  { value: 'ATL', label: 'Atlanta (ATL)', sublabel: 'Hartsfield-Jackson International' },
-  { value: 'DFW', label: 'Dallas (DFW)', sublabel: 'Dallas/Fort Worth International' },
-];
-
+type TripType = 'roundtrip' | 'oneway' | 'multicity';
 
 const PlaneIcon = () => (
   <svg className="w-5 h-5" fill="none" stroke="currentColor" viewBox="0 0 24 24">
@@ -32,9 +22,11 @@ const LocationIcon = () => (
 
 export default function AirlineHero() {
   const router = useRouter();
+  const [tripType, setTripType] = useState<TripType>('roundtrip');
   const [origin, setOrigin] = useState('');
   const [destination, setDestination] = useState('');
   const [departDate, setDepartDate] = useState('');
+  const [returnDate, setReturnDate] = useState('');
   const [passengers, setPassengers] = useState({ adults: 1, children: 0, infants: 0 });
 
   const handleSearch = (e: FormEvent) => {
@@ -48,6 +40,8 @@ export default function AirlineHero() {
 
     if (origin) params.set('origin', origin);
     if (destination) params.set('destination', destination);
+    if (tripType !== 'roundtrip') params.set('tripType', tripType);
+    if (returnDate && tripType === 'roundtrip') params.set('returnDate', returnDate);
 
     params.set('adults', passengers.adults.toString());
     params.set('children', passengers.children.toString());
@@ -72,15 +66,28 @@ export default function AirlineHero() {
 
         <div className="search-form">
           <form onSubmit={handleSearch}>
-            <div className="grid grid-cols-1 sm:grid-cols-2 lg:grid-cols-3 gap-4">
+            <div className="mb-6">
+              <RadioGroup
+                name="tripType"
+                value={tripType}
+                onChange={setTripType}
+                options={[
+                  { value: 'roundtrip', label: 'Round-trip' },
+                  { value: 'oneway', label: 'One-way' },
+                  { value: 'multicity', label: 'Multi-city' },
+                ]}
+              />
+            </div>
+
+            <div className="grid grid-cols-1 sm:grid-cols-2 lg:grid-cols-4 gap-4">
               <div>
                 <Label htmlFor="origin">From</Label>
-                <SelectDropdown
+                <Input
+                  type="text"
                   id="origin"
                   value={origin}
-                  onChange={setOrigin}
-                  options={CITIES}
-                  placeholder="Select origin"
+                  onChange={(e) => setOrigin(e.target.value)}
+                  placeholder="Airport or city"
                   icon={<PlaneIcon />}
                   required
                 />
@@ -88,12 +95,12 @@ export default function AirlineHero() {
 
               <div>
                 <Label htmlFor="destination">To</Label>
-                <SelectDropdown
+                <Input
+                  type="text"
                   id="destination"
                   value={destination}
-                  onChange={setDestination}
-                  options={CITIES}
-                  placeholder="Select destination"
+                  onChange={(e) => setDestination(e.target.value)}
+                  placeholder="Airport or city"
                   icon={<LocationIcon />}
                   required
                 />
@@ -108,6 +115,20 @@ export default function AirlineHero() {
                   required
                 />
               </div>
+
+              <div>
+                <Label htmlFor="returnDate">Return</Label>
+                {tripType === 'roundtrip' ? (
+                  <DateInput
+                    id="returnDate"
+                    value={returnDate}
+                    onChange={(e) => setReturnDate(e.target.value)}
+                    required
+                  />
+                ) : (
+                  <DateInput id="returnDate" disabled />
+                )}
+              </div>
             </div>
 
             <div className="grid grid-cols-4 sm:grid-cols-3 lg:grid-cols-4 gap-4 mt-4">

web/src/components/feats/hotel/HotelCard.tsx

@@ -21,7 +21,7 @@ const AmenityIcon = ({ name }: { name: string }) => {
         breakfast: 'Breakfast',
         spa: 'Spa',
     };
-    return <span className="feature-tag">{iconMap[name.toLowerCase()] || name.replaceAll("_", " ")}</span>;
+    return <span className="feature-tag">{iconMap[name.toLowerCase()] || name}</span>;
 };
 
 export default function HotelCard({ hotel }: { hotel: Hotel }) {
@@ -47,31 +47,18 @@ export default function HotelCard({ hotel }: { hotel: Hotel }) {
         window.location.href = `/hotel/products/${hotel.id}`;
     };
 
-    const imageUrl = `https://images.unsplash.com/photo-1551882547-ff40c63fe5fa?w=400&h=300&fit=crop`;
-
     return (
         <div
             className="hotel-card cursor-pointer"
             onClick={handleCardClick}
         >
-            <div className="hotel-image relative overflow-hidden">
-                <img
-                    src={imageUrl}
-                    alt={hotel.name}
-                    className="w-full h-full object-cover"
-                    onError={(e) => {
-                        e.currentTarget.style.display = 'none';
-                        const fallback = e.currentTarget.nextElementSibling as HTMLElement;
-                        if (fallback) fallback.style.display = 'flex';
-                    }}
-                />
-                <div className="absolute inset-0 bg-gray-200 flex items-center justify-center" style={{ display: 'none' }}>
-                    <span className="text-gray-400 text-sm">Image</span>
-                </div>
+            <div className="hotel-image bg-gray-200 flex items-center justify-center">
+                <span className="text-gray-400 text-sm">Image</span>
             </div>
 
             <div className="hotel-info">
                 <h3 ref={titleRef} className="hotel-name">{hotel.name}</h3>
+                <div className="hotel-location text-sm mb-2">{hotel.roomType}</div>
                 <div className="text-sm text-[var(--text-secondary)] mb-2">
                     {hotel.checkIn} - {hotel.checkOut}
                 </div>
@@ -80,6 +67,9 @@ export default function HotelCard({ hotel }: { hotel: Hotel }) {
                         <AmenityIcon key={a} name={a} />
                     ))}
                 </div>
+                {hotel.refundable && (
+                    <div className="free-cancellation mt-2">Free cancellation</div>
+                )}
             </div>
 
             <div className="hotel-pricing">

web/src/components/feats/hotel/HotelDetails.tsx

@@ -1,8 +1,6 @@
 'use client';
 
-import { useState, useEffect } from 'react';
 import type { Hotel } from '@/lib/hotel-utils';
-import PriceDisplay from '@/components/ui/PriceDisplay';
 
 interface HotelDetailsProps {
   product: Hotel;
@@ -10,63 +8,19 @@ interface HotelDetailsProps {
   addedToCart: boolean;
 }
 
-const PriceTotalDisplay = ({ productId, nights }: { productId: string; nights: number }) => {
-  const [price, setPrice] = useState<number | null>(null);
-
-  useEffect(() => {
-    const fetchPrice = async () => {
-      try {
-        const sessionRes = await fetch('/api/session');
-        const sessionData = await sessionRes.json();
-        const params = new URLSearchParams({
-          productId,
-          sessionId: sessionData.sessionId || '',
-          experimentId: sessionData.experimentId || '',
-        });
-        const res = await fetch(`/api/pricing?${params.toString()}`);
-        const data = await res.json();
-        setPrice(data.price);
-      } catch (err) {
-        console.error('failed to fetch price for total:', err);
-      }
-    };
-    fetchPrice();
-  }, [productId]);
-
-  if (!price) return <span className="text-4xl font-bold text-gray-900">Loading...</span>;
-
-  return (
-    <span className="text-4xl font-bold text-gray-900">
-      ${(price * nights).toFixed(2)}
-    </span>
-  );
-};
-
 export default function HotelDetails({ product, onAddToCart, addedToCart }: HotelDetailsProps) {
-  const imageUrl = `https://images.unsplash.com/photo-1566073771259-6a8506099945?w=800&h=600&fit=crop`;
-
   return (
     <div className="w-full flex flex-col lg:flex-row gap-12 py-8">
-      <div className="w-full lg:w-1/2 rounded-lg aspect-[4/3] overflow-hidden shrink-0">
-        <img
-          src={imageUrl}
-          alt={product.name}
-          className="w-full h-full object-cover"
-          onError={(e) => {
-            e.currentTarget.style.display = 'none';
-            if (e.currentTarget.nextElementSibling) {
-              (e.currentTarget.nextElementSibling as HTMLElement).style.display = 'flex';
-            }
-          }}
-        />
-        <div className="w-full h-full bg-gray-100 rounded-lg flex items-center justify-center" style={{ display: 'none' }}>
-          <span className="text-gray-400 text-lg font-medium">Hotel Image</span>
-        </div>
+      {/* Image Section - Larger and cleaner */}
+      <div className="w-full lg:w-1/2 bg-gray-100 rounded-lg aspect-[4/3] flex items-center justify-center shrink-0">
+        <span className="text-gray-400 text-lg font-medium">Hotel Image</span>
       </div>
 
+      {/* Details Section - Full height/width usage */}
       <div className="flex-1 flex flex-col">
         <div className="border-b pb-6 mb-6">
           <h1 className="text-4xl font-bold text-gray-900 mb-2">{product.name}</h1>
+          <p className="text-xl text-gray-500">{product.roomType}</p>
         </div>
 
         <div className="grid grid-cols-2 gap-8 mb-8">
@@ -85,17 +39,24 @@ export default function HotelDetails({ product, onAddToCart, addedToCart }: Hote
           <div className="flex flex-wrap gap-3">
             {product.amenities.map(a => (
               <span key={a} className="px-3 py-1.5 bg-gray-100 text-gray-700 rounded-md text-sm font-medium">
-                  {a.replaceAll('_', ' ')}
+                {a}
               </span>
             ))}
           </div>
         </div>
 
+        {product.refundable && (
+          <div className="mb-8 p-4 bg-green-50 text-green-800 rounded-md inline-block">
+            <span className="font-medium">Free cancellation available</span>
+          </div>
+        )}
+
         <div className="mt-auto pt-6 border-t flex items-center justify-between">
           <div>
-            <p className="text-sm text-gray-500 mb-1">Price per night</p>
-            <div className="mb-3">
-              <PriceDisplay productId={product.id} className="!text-2xl" />
+            <p className="text-sm text-gray-500 mb-1">Total for {product.nights} night{product.nights > 1 ? 's' : ''}</p>
+            <div className="flex items-baseline gap-2">
+              <span className="text-4xl font-bold text-gray-900">${product.pricePerNight * product.nights}</span>
+              <span className="text-gray-500">/ {product.nights} nights</span>
             </div>
           </div>
 

web/src/components/ui/DateInput.tsx

@@ -1,29 +1,7 @@
-import { InputHTMLAttributes, useMemo } from 'react';
+import { InputHTMLAttributes } from 'react';
 
 interface DateInpProps extends Omit<InputHTMLAttributes<HTMLInputElement>, 'type'> {}
 
 export default function DateInput({ className = '', ...props }: DateInpProps) {
-  const { minDate, maxDate } = useMemo(() => {
-    const today = new Date();
-    const tomorrow = new Date(today);
-    tomorrow.setDate(today.getDate() + 1);
-
-    const tenDaysOut = new Date(tomorrow);
-    tenDaysOut.setDate(tomorrow.getDate() + 9); // tomorrow + 9 = 10 days total
-
-    return {
-      minDate: tomorrow.toISOString().split('T')[0],
-      maxDate: tenDaysOut.toISOString().split('T')[0]
-    };
-  }, []);
-
-  return (
-    <input
-      type="date"
-      className={`input-field ${className}`.trim()}
-      min={minDate}
-      max={maxDate}
-      {...props}
-    />
-  );
+  return <input type="date" className={`input-field ${className}`.trim()} {...props} />;
 }

web/src/components/ui/Navigation.tsx

@@ -20,7 +20,7 @@ const NavLink = ({ href, children }: { href: string; children: React.ReactNode }
       href={href}
       className={`px-4 py-2 rounded-md transition-colors ${
         isActive
-          ? 'bg-[var(--accent-primary)] font-semibold'
+          ? 'bg-[var(--accent-primary)] text-white font-semibold'
           : 'hover:bg-[var(--accent-primary-light)] text-[var(--text-primary)]'
       }`}
     >
@@ -37,7 +37,9 @@ export default function Navigation() {
           <div className="flex items-center space-x-1">
             <NavLink href="/">Home</NavLink>
             <NavLink href="/products">Products</NavLink>
+            <NavLink href="/search">Search</NavLink>
             <NavLink href="/cart">Cart</NavLink>
+            <NavLink href="/checkout">Checkout</NavLink>
           </div>
         </div>
       </div>

web/src/components/ui/SelectDropdown.tsx

@@ -1,119 +0,0 @@
-'use client';
-
-import { useState, useRef, useEffect, ReactNode } from 'react';
-
-export interface SelectOption {
-  value: string;
-  label: string;
-  sublabel?: string;
-}
-
-interface SelectDropdownProps {
-  value: string;
-  onChange: (value: string) => void;
-  options: SelectOption[];
-  placeholder?: string;
-  icon?: ReactNode;
-  required?: boolean;
-  id?: string;
-}
-
-export default function SelectDropdown({
-  value,
-  onChange,
-  options,
-  placeholder = 'Select...',
-  icon,
-  required,
-  id,
-}: SelectDropdownProps) {
-  const [open, setOpen] = useState(false);
-  const [filter, setFilter] = useState('');
-  const ref = useRef<HTMLDivElement>(null);
-  const inputRef = useRef<HTMLInputElement>(null);
-
-  useEffect(() => {
-    const handleClick = (e: MouseEvent) => {
-      if (ref.current && !ref.current.contains(e.target as Node)) {
-        setOpen(false);
-        setFilter('');
-      }
-    };
-    document.addEventListener('mousedown', handleClick);
-    return () => document.removeEventListener('mousedown', handleClick);
-  }, []);
-
-  const selectedOption = options.find((o) => o.value === value);
-  const filtered = options.filter(
-    (o) =>
-      o.label.toLowerCase().includes(filter.toLowerCase()) ||
-      o.value.toLowerCase().includes(filter.toLowerCase()) ||
-      o.sublabel?.toLowerCase().includes(filter.toLowerCase())
-  );
-
-  const handleSelect = (opt: SelectOption) => {
-    onChange(opt.value);
-    setOpen(false);
-    setFilter('');
-  };
-
-  return (
-    <div className="relative" ref={ref}>
-      <div
-        className="input-field flex items-center gap-2 cursor-pointer box-border"
-        onClick={() => {
-          setOpen(true);
-          setTimeout(() => inputRef.current?.focus(), 0);
-        }}
-      >
-        {icon && <span className="text-[var(--text-secondary)]">{icon}</span>}
-        {open ? (
-          <input
-            ref={inputRef}
-            type="text"
-            id={id}
-            value={filter}
-            onChange={(e) => setFilter(e.target.value)}
-            placeholder={placeholder}
-            className="flex-1 bg-transparent outline-none text-sm text-[var(--text-primary)]"
-          />
-        ) : (
-          <span className={`flex-1 text-sm ${value ? 'text-[var(--text-primary)]' : 'text-[var(--text-secondary)]'}`}>
-            {selectedOption ? selectedOption.label : placeholder}
-          </span>
-        )}
-        <svg
-          className={`w-4 h-4 text-[var(--text-secondary)] transition-transform ${open ? 'rotate-180' : ''}`}
-          fill="none"
-          stroke="currentColor"
-          viewBox="0 0 24 24"
-        >
-          <path strokeLinecap="round" strokeLinejoin="round" strokeWidth={2} d="M19 9l-7 7-7-7" />
-        </svg>
-      </div>
-      {open && (
-        <div className="absolute z-20 mt-1 w-full bg-[var(--bg-primary)] border-2 border-[var(--accent-primary)] rounded-md shadow-lg max-h-60 overflow-y-auto">
-          {filtered.length === 0 ? (
-            <div className="px-4 py-3 text-sm text-[var(--text-secondary)]">No results</div>
-          ) : (
-            filtered.map((opt) => (
-              <div
-                key={opt.value}
-                onClick={() => handleSelect(opt)}
-                className={`px-4 py-2 cursor-pointer transition-colors hover:bg-[var(--accent-primary-light)] ${
-                  opt.value === value ? 'bg-[var(--accent-primary-light)]' : ''
-                }`}
-              >
-                <div className="text-sm font-medium text-[var(--text-primary)]">{opt.label}</div>
-                {opt.sublabel && <div className="text-xs text-[var(--text-secondary)]">{opt.sublabel}</div>}
-              </div>
-            ))
-          )}
-        </div>
-      )}
-      {required && !value && (
-        <input type="text" required className="sr-only" tabIndex={-1} value="" onChange={() => {}} />
-      )}
-    </div>
-  );
-}

web/src/components/ui/index.ts

@@ -5,5 +5,3 @@ export { default as DateInput } from './DateInput';
 export { default as RadioGroup } from './RadioGroup';
 export { default as Dropdown, DropdownCounter } from './Dropdown';
 export { default as Navigation } from './Navigation';
-export { default as SelectDropdown } from './SelectDropdown';
-export type { SelectOption } from './SelectDropdown';

web/src/lib/config.ts

@@ -16,7 +16,7 @@ const envSchema = z.object({
 // parse and validate env at module load, fail fast with descriptive errors
 const parseEnv = (): Env => {
     const result = envSchema.safeParse({
-        STORE_MODE: process.env.NEXT_PUBLIC_STORE_MODE || process.env.STORE_MODE,
+        STORE_MODE: process.env.STORE_MODE,
         NEXT_PUBLIC_API_BASE: process.env.NEXT_PUBLIC_API_BASE,
         NEXT_PUBLIC_APP_ENV: process.env.NEXT_PUBLIC_APP_ENV,
     });

web/src/lib/hotel-utils.ts

@@ -21,6 +21,7 @@ export interface Hotel {
   checkOut: string;
   dateIndex: number;
   amenities: string[];
+  refundable: boolean;
   pricePerNight: number;
   nights: number;
 }
@@ -29,37 +30,19 @@ const EPOCH = new Date(0);
 
 export const transformProduct = (p: HotelProduct): Hotel => {
   const { id, room_type, date_index, metadata } = p;
-
-  // DB stores date_index as days since epoch
-  // but if value is small (<1000), treat as days from today for backward compat
-  let checkIn: Date;
-  if (date_index < 1000) {
-    // legacy: treat as offset from today
-    const today = new Date();
-    today.setHours(0, 0, 0, 0);
-    checkIn = new Date(today.getTime() + date_index * 86400000);
-  } else {
-    // proper: days since epoch
-    checkIn = new Date(EPOCH.getTime() + date_index * 86400000);
-  }
-
+  const checkIn = new Date(EPOCH.getTime() + date_index * 86400000);
   const nights = 1;
   const checkOut = new Date(checkIn.getTime() + nights * 86400000);
 
-  const formatOpts: Intl.DateTimeFormatOptions = {
-    month: 'short',
-    day: 'numeric',
-    year: checkIn.getFullYear() !== new Date().getFullYear() ? 'numeric' : undefined
-  };
-
   return {
     id,
     name: metadata?.name || room_type,
     roomType: room_type,
-    checkIn: checkIn.toLocaleDateString('en-US', formatOpts),
-    checkOut: checkOut.toLocaleDateString('en-US', formatOpts),
+    checkIn: checkIn.toLocaleDateString('en-US', { month: 'short', day: 'numeric' }),
+    checkOut: checkOut.toLocaleDateString('en-US', { month: 'short', day: 'numeric' }),
     dateIndex: date_index,
     amenities: metadata?.amenities || [],
+    refundable: metadata?.refundable || false,
     pricePerNight: metadata?.base_price || 100,
     nights,
   };

web/src/styles/airline.css

@@ -278,8 +278,6 @@
   padding: 12px;
   transition: border-color 0.2s ease;
   width: 100%;
-  min-height: 48px;
-  box-sizing: border-box;
 }
 
 [data-mode="airline"] .input-field:focus {