chore: clean up definition of composite class of providers

README.md

@@ -1,12 +1,8 @@
-<img width="200" align="left" src="https://github.com/user-attachments/assets/d148b00d-e9f9-4280-89cc-0cc866e17251" />
 
-### PHANTOM
+<img width="1952" height="2176" alt="nobody_knows" src="https://github.com/user-attachments/assets/d148b00d-e9f9-4280-89cc-0cc866e17251" />
 
 [![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/
 

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:

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/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/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/pipelines.py

@@ -2,7 +2,6 @@ from sklearn.pipeline import Pipeline
 import pandas as pd
 from procesing.context import PipelineContext
 from procesing.providers import SupabaseProvider, BackendAPIProvider
-import os
 from procesing.steps import (
     FetchInteractionsStep,
     FetchPriceLogsStep,
@@ -13,13 +12,11 @@ from procesing.steps import (
     ChunkByTimeWindowStep,
     ComputeDemandForChunksStep,
     AggregatePriceLogsStep,
+    # BuildStateSpaceStep,
     FitPricingFunctionStep,
     PredictPricesStep,
     ComputeDemandStep,
-    JoinProductFeaturesStep,
-    ExtractSessionFeaturesStep,
-    JoinLabelsStep,
-    ValidateDataStep,
+    JoinProductFeaturesStep
 )
 from procesing.pricers import SimpleSurgePricer
 
@@ -109,66 +106,33 @@ def full_pipeline(context: PipelineContext,
     return product_features_df, optimal_prices_df
 
 
-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
-
-
 
 
 if __name__ == '__main__':
 
-    class ExperimentsProvider(SupabaseProvider, BackendAPIProvider):
+    class Provider(SupabaseProvider, BackendAPIProvider):
+        def __init__(self, backend_url: str):
+            SupabaseProvider.__init__(self)
+            BackendAPIProvider.__init__(self, backend_url=backend_url)
+
+
+    class HistoricalProvider(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()
+            path = "/home/velocitatem/Documents/Projects/PHANTOM/experiments/collected_data/858c61ab-0a7f-4595-ae49-33f4365517b9/"
+            interactions_file = "messages(2).json"
+            prices_file = "messages(3).json"
 
-            files = {"user-interactions": "int.json", "price-logs": "price.json"}
-            file_to_read = files.get(topic, files["user-interactions"])
-            frames = []
+            data = pd.read_json(path + (interactions_file if topic == "user-interactions" else prices_file))
+            data = [r['payload'] for r in data['value'].to_list()]
+            data = pd.DataFrame(data)
+            return data
 
-            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}")
 
-            return pd.concat(frames, ignore_index=True) if frames else pd.DataFrame()
+    # example run
+    context = PipelineContext(
+        provider=HistoricalProvider(),
+        store_mode='airline',
+    )
 
-    # 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")
+    product_features, prices = full_pipeline(context)
+    print(prices.to_string())

experiments/procesing/steps/__init__.py

@@ -6,11 +6,7 @@ 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.session import ExtractSessionFeaturesStep, _extract_features_for_session
 
 __all__ = [
     'BaseContextStep',
@@ -29,11 +25,5 @@ __all__ = [
     'FitPricingFunctionStep',
     'PredictPricesStep',
     'ExtractSessionFeaturesStep',
-    'JoinLabelsStep',
-    'ValidateDataStep',
-    'TemporalFeatureStep',
-    'BehavioralFeatureStep',
-    'ProductFeatureStep',
-    'UserAgentFeatureStep',
     '_extract_features_for_session',
 ]

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,12 +7,12 @@ 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')

experiments/procesing/steps/session.py

@@ -1,261 +1,159 @@
 """
-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 _extract_features_for_session(session_df: pd.DataFrame, session_timeout_sec: float = 900) -> Dict[str, Any]:
+    """Compute features for single session.
+
+    Args:
+        session_df: interaction events for this session
+        session_timeout_sec: max gap between events before resetting duration (default 900s = 15min)
+    """
+    features = {}
+
+    # basic counts
+    features['total_interactions'] = len(session_df)
+
+    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)
+
+    # 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['product_view_depth'] = 0
+
+    # temporal features with session timeout logic
+    if 'ts' in session_df.columns:
+        timestamps = session_df['ts'].sort_values()
+
+        # compute active duration considering timeout gaps
+        if len(timestamps) > 1:
+            time_diffs = timestamps.diff().dropna().dt.total_seconds()
+            # only count gaps shorter than timeout towards active session duration
+            active_diffs = time_diffs[time_diffs <= session_timeout_sec]
+            features['session_duration_sec'] = active_diffs.sum() if len(active_diffs) > 0 else 0.0
+
+            features['avg_time_between_events'] = time_diffs.mean()
+            features['std_time_between_events'] = time_diffs.std()
+        else:
+            features['session_duration_sec'] = 0.0
+            features['avg_time_between_events'] = 0.0
+            features['std_time_between_events'] = 0.0
+
+        if features['session_duration_sec'] > 0:
+            features['interaction_velocity'] = (features['total_interactions'] / features['session_duration_sec']) * 60
+        else:
+            features['interaction_velocity'] = 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
 
 
-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'
+def _apply_to_slice(df: pd.DataFrame) -> pd.DataFrame:
+    """Apply feature extraction to sliding window of interactions."""
+    # add columns of all features at each step
+    new_cols = ["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"]
+    for col in new_cols: df[col] = np.nan
+    for idx in range(1, len(df) + 1):
+        features = _extract_features_for_session(df.iloc[:idx])
+        # fillna kinda meh
+        features = { k: (v if not pd.isna(v) else 0.0) for k, v in features.items() }
+        for col in new_cols:
+            df.at[df.index[idx - 1], col] = features[col]
+        #print(f"Processed {idx}/{len(df)} events for session {df['sessionId'].iloc[0]}")
+    return df
+
+class BuildStateSpaceStep(BaseContextStep):
+    """
+    Build state space representation S_t from session features.
+
+    Input: session_features DataFrame
+    Output: state_space_df DataFrame with S_t vectors
+    """
+
+    def transform(self, rich_dataset: pd.DataFrame) -> pd.DataFrame:
+        # check if features are present
+        required_cols = ["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 not all(col in rich_dataset.columns for col in required_cols):
+            raise ValueError("Missing required columns for feature extraction.")
+        if rich_dataset.empty:
+            return pd.DataFrame()
 
 
-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
+        # For simplicity, we return as is
+        return rich_dataset.copy()
 
 
-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: interactions_df with added session feature columns
     """
 
-    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'):
+            new_slice = _apply_to_slice(session_df.sort_values('ts'))
+            session_features.append(new_slice)
 
-        # 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 result
+        return pd.concat(session_features, ignore_index=True)
 
 
-class JoinLabelsStep(BaseContextStep):
+
+class FilterSessionInteractionsStep(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
+    Filter interactions DataFrame to specific session.
+
+    Input: (interactions_df, session_id)
+    Output: interactions_df filtered to session_id
     """
 
-    def transform(self, X : tuple) -> pd.DataFrame:
-        data = X;
-        if isinstance(data, tuple):
-            features_df, experiments_df = data
-        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()
-
-        if features_df.empty:
-            return features_df
-        if experiments_df.empty:
-            features_df['is_agent'] = np.nan
-            return features_df
-
-        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']
-
-        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')
-
-
-class ValidateDataStep(BaseContextStep):
-    """
-    Data quality checks before training.
-    Input: df
-    Output: df (unchanged, but logs validation report to context)
-    """
-    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

@@ -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