fix: adding date utils

* a simp0le scaffold

* feature: simple npm setup

* feature: testing setup and dummy scenarios

* chore: dumping kafak just via backend

* chore: dcleaning gitignore

* features: boilerplate fixtures and stuff

* test: extra tests

* chore: update the test suite to be callable via makefile

* chore: cleaning

* chore: updating interactions setup

* small cleaning

* chore: cleaning shitty code

.gitignore

@@ -11,3 +11,6 @@ paper/src/bib/auto
 experiments/airflow/logs/*
 experiments/airflow/logs/scheduler/
 experiments/airflow/logs/dag_processor_manager/
+tests/e2e/node_modules/**
+**/auto/*.el
+*.old

Makefile

@@ -11,46 +11,74 @@ PYTEST    := $(VENV)/bin/pytest
 
 .DEFAULT_GOAL := help
 
-all: pdf
+.PHONY: help
-
+help:
-run.webapp:
+	@echo "pdf.build pdf.watch pdf.clean | test.backend test.e2e test.all | web.dev | install | stats.lines"
-	@cd web && npm install && npm run dev
 
 $(BUILDDIR):
 	mkdir -p paper/$(BUILDDIR)
 
-pdf: $(BUILDDIR)
+.PHONY: pdf.build
-	@echo "Concatenating source code..."
+pdf.build: $(BUILDDIR)
 	@bash paper/concat_code.sh
 	@cd $(SRCDIR) && \
 	$(LATEXMK) -pdf -jobname=$(JOBNAME) \
 		-interaction=nonstopmode -file-line-error \
 		-outdir=../$(BUILDDIR) $(TEX)
 
-watch: $(BUILDDIR)
+.PHONY: pdf.watch
+pdf.watch: $(BUILDDIR)
 	@cd $(SRCDIR) && \
 	$(LATEXMK) -pvc -pdf -jobname=$(JOBNAME) \
 		-interaction=nonstopmode -file-line-error \
 		-r ../.latexmkrc \
 		-outdir=../$(BUILDDIR) $(TEX)
 
-clean:
+.PHONY: pdf.clean
+pdf.clean:
 	@cd $(SRCDIR) && \
 	$(LATEXMK) -C -jobname=$(JOBNAME) -outdir=../$(BUILDDIR) || true
 	rm -rf paper/$(BUILDDIR)/*
 
+.PHONY: test.backend
+test.backend: $(VENV)
+	$(PYTEST) -v
+
+.PHONY: test.e2e
+test.e2e:
+	@cd tests/e2e && npm install
+	@cd tests/e2e && npx playwright install chromium
+	@test -f tests/e2e/.env || cp tests/e2e/.env.example tests/e2e/.env
+	@timeout 30 bash -c 'until curl -sf http://localhost:5000/health > /dev/null 2>&1; do sleep 1; done' || (echo "Backend not ready" && exit 1)
+	@timeout 30 bash -c 'until curl -sf http://localhost:3000 > /dev/null 2>&1; do sleep 1; done' || (echo "Web app not ready" && exit 1)
+	@timeout 30 bash -c 'until curl -sf http://localhost:8085/health > /dev/null 2>&1; do sleep 1; done' || (echo "Airflow not ready" && exit 1)
+	@cd tests/e2e && npm test
+
+.PHONY: test.all
+test.all: test.backend test.e2e
+
+.PHONY: web.dev
+web.dev:
+	@cd web && npm install && npm run dev
+
 $(VENV):
 	python3 -m venv $(VENV)
 	$(PIP) install --upgrade pip
 
+.PHONY: install
 install: $(VENV)
 	$(PIP) install -r requirements.txt
 
-test: $(VENV)
+.PHONY: stats.lines
-	$(PYTEST) -v
+stats.lines:
-
-count-lines:
 	@find . \( -path '*/node_modules' -o -path '*/.venv' -o -path '*/venv' \) -prune -o \
 	\( -name "*.ts" -o -name "*.py" \) -type f -print0 | xargs -0 cat | wc -l
 
-.PHONY: all pdf clean watch run.webapp install test
+.PHONY: pdf clean watch run.webapp test count-lines all
+pdf: pdf.build
+clean: pdf.clean
+watch: pdf.watch
+run.webapp: web.dev
+test: test.backend
+count-lines: stats.lines
+all: pdf.build

README.md

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

backend/provider/app.py

@@ -47,53 +47,52 @@ def health() -> dict:
 
 @app.get("/api/{mode}/price/{productId}", response_model=PriceResponse)
 def get_price(mode: Literal['hotel', 'airline'], productId: str, sessionId: Optional[str] = Query(None), experimentId: Optional[str] = Query(None)):
+    """
+    THIS is the fast lookup service (mechanism).
+    Priority: session-keyed price > global optimal price > base price
+    """
     product = supabase.table(f'{mode}_products').select("metadata").eq('id', productId).execute().data[0]
     if not product: raise HTTPException(404, f"Product {productId} not found")
 
     metadata = product['metadata']
     base_price = metadata.get('base_price', 100.0)
 
-    # fetch pre-computed prices from registry
+    # PRIORITY 1: session-aware price (computed by Airflow worker)
+    if sessionId:
+        session_price = registry.get_session_price(sessionId, productId)
+        if session_price is not None:
+            return PriceResponse(
+                productId=productId,
+                price=session_price,
+                base_price=base_price,
+                markup=session_price/base_price,
+                elasticity=None,
+                model_version='session-aware'
+            )
+
+    # PRIORITY 2: global pre-computed prices (surge pricing)
     prices_df = registry.get_prices('latest')
-    elasticity_df = registry.get_elasticity('latest')
+    if prices_df is not None:
-
-    if prices_df is None:
-        # fallback: no pre-computed prices available
-        return PriceResponse(
-            productId=productId,
-            price=base_price,
-            base_price=base_price,
-            markup=1.0,
-            elasticity=None
-        )
-
-    # lookup pre-computed price for this product
         product_price_row = prices_df[prices_df['productId'] == productId]
-    if product_price_row.empty:
+        if not product_price_row.empty:
-        # product not in pre-computed prices, fallback to base
+            optimal_price = float(product_price_row['optimal_price'].iloc[0])
-        return PriceResponse(
-            productId=productId,
-            price=base_price,
-            base_price=base_price,
-            markup=1.0,
-            elasticity=None
-        )
-
-    optimal_price = float(product_price_row['optimal_price'].iloc[0]) # TODO: use optimal_price everywhere as  aresult
-
-    # get elasticity if available
-    product_elasticity = None
-    if elasticity_df is not None:
-        product_elasticity_row = elasticity_df[elasticity_df['productId'] == productId]
-        if not product_elasticity_row.empty:
-            product_elasticity = float(product_elasticity_row['elasticity'].iloc[0])
-
             return PriceResponse(
                 productId=productId,
                 price=optimal_price,
                 base_price=base_price,
                 markup=optimal_price/base_price,
-        elasticity=product_elasticity
+                elasticity=None,
+                model_version='surge'
+            )
+
+    # PRIORITY 3: fallback to base price
+    return PriceResponse(
+        productId=productId,
+        price=base_price,
+        base_price=base_price,
+        markup=1.0,
+        elasticity=None,
+        model_version='base'
     )
 
 @app.get("/models")

backend/server/app.py

@@ -198,12 +198,16 @@ def dump_logs(
             auto_offset_reset='earliest',
             enable_auto_commit=False,
             value_deserializer=lambda x: json.loads(x.decode('utf-8')),
-            consumer_timeout_ms=5000
+            consumer_timeout_ms=30000,
+            fetch_max_wait_ms=10000,
+            max_poll_records=1000
         )
 
         events = []
         for msg in consumer:
             events.append(msg.value)
+            if last_n and len(events) >= last_n * 2:
+                break
 
         consumer.close()
 

docker-compose.yml

@@ -1,4 +1,24 @@
 services:
+  tensorboard-rl:
+    image: tensorflow/tensorflow:latest
+    container_name: "PHANTOM-tensorboard-rl"
+    ports:
+      - "6007:6006"
+    volumes:
+      - ./sim/rl/runs:/logs
+    command: tensorboard --logdir=/logs --host=0.0.0.0 --port=6006
+    restart: unless-stopped
+
+  tensorboard-ml:
+    image: tensorflow/tensorflow:latest
+    container_name: "PHANTOM-tensorboard-ml"
+    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:
@@ -92,11 +112,14 @@ services:
     depends_on:
       - postgres
     environment:
-      - AIRFLOW__CORE__EXECUTOR=SequentialExecutor
+      - AIRFLOW__CORE__EXECUTOR=LocalExecutor
       - AIRFLOW__DATABASE__SQL_ALCHEMY_CONN=postgresql+psycopg2://airflow:airflow@postgres/airflow
       - AIRFLOW__CORE__FERNET_KEY=${AIRFLOW_FERNET_KEY}
       - AIRFLOW__CORE__LOAD_EXAMPLES=false
       - AIRFLOW__CORE__ENABLE_XCOM_PICKLING=true
+      - AIRFLOW__CORE__PARALLELISM=16
+      - AIRFLOW__CORE__DAG_CONCURRENCY=8
+      - AIRFLOW__CORE__MAX_ACTIVE_RUNS_PER_DAG=4
       - _AIRFLOW_DB_MIGRATE=true
       - _AIRFLOW_WWW_USER_CREATE=true
       - _AIRFLOW_WWW_USER_USERNAME=admin
@@ -116,14 +139,20 @@ services:
       - airflow-init
       - redis
     environment:
-      - AIRFLOW__CORE__EXECUTOR=SequentialExecutor
+      - AIRFLOW__CORE__EXECUTOR=LocalExecutor
       - AIRFLOW__DATABASE__SQL_ALCHEMY_CONN=postgresql+psycopg2://airflow:airflow@postgres/airflow
       - AIRFLOW__CORE__FERNET_KEY=${AIRFLOW_FERNET_KEY}
       - AIRFLOW__CORE__DAGS_ARE_PAUSED_AT_CREATION=true
       - AIRFLOW__CORE__LOAD_EXAMPLES=false
       - AIRFLOW__CORE__ENABLE_XCOM_PICKLING=true
+      - AIRFLOW__CORE__PARALLELISM=16
+      - AIRFLOW__CORE__DAG_CONCURRENCY=8
+      - AIRFLOW__CORE__MAX_ACTIVE_RUNS_PER_DAG=4
+      - AIRFLOW__SCHEDULER__MIN_FILE_PROCESS_INTERVAL=30
+      - AIRFLOW__SCHEDULER__DAG_DIR_LIST_INTERVAL=60
       - AIRFLOW__WEBSERVER__EXPOSE_CONFIG=true
       - AIRFLOW__WEBSERVER__SECRET_KEY=${AIRFLOW_SECRET_KEY}
+      - AIRFLOW__API__AUTH_BACKENDS=airflow.api.auth.backend.basic_auth
       - KAFKA_HOST=kafka
       - KAFKA_PORT=29092
       - BACKEND_URL=http://backend:5000
@@ -153,13 +182,20 @@ services:
       redis:
         condition: service_started
     environment:
-      - AIRFLOW__CORE__EXECUTOR=SequentialExecutor
+      - AIRFLOW__CORE__EXECUTOR=LocalExecutor
       - AIRFLOW__DATABASE__SQL_ALCHEMY_CONN=postgresql+psycopg2://airflow:airflow@postgres/airflow
       - AIRFLOW__CORE__FERNET_KEY=${AIRFLOW_FERNET_KEY}
       - AIRFLOW__CORE__DAGS_ARE_PAUSED_AT_CREATION=true
       - AIRFLOW__CORE__LOAD_EXAMPLES=false
       - AIRFLOW__CORE__ENABLE_XCOM_PICKLING=true
+      - AIRFLOW__CORE__PARALLELISM=16
+      - AIRFLOW__CORE__DAG_CONCURRENCY=8
+      - AIRFLOW__CORE__MAX_ACTIVE_RUNS_PER_DAG=4
+      - AIRFLOW__SCHEDULER__MIN_FILE_PROCESS_INTERVAL=30
+      - AIRFLOW__SCHEDULER__DAG_DIR_LIST_INTERVAL=60
+      - AIRFLOW__SCHEDULER__PARSING_PROCESSES=2
       - AIRFLOW__WEBSERVER__SECRET_KEY=${AIRFLOW_SECRET_KEY}
+      - AIRFLOW__API__AUTH_BACKENDS=airflow.api.auth.backend.basic_auth
       - KAFKA_HOST=kafka
       - KAFKA_PORT=29092
       - BACKEND_URL=http://backend:5000

experiments/airflow/dags/ml_training_pipeline.py

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

experiments/airflow/dags/surge_pricing_factory.py

@@ -1,3 +1,4 @@
+from pandas.core.algorithms import factorize_array
 from airflow import DAG
 from airflow.operators.python import PythonOperator
 from airflow.utils.dates import days_ago
@@ -208,3 +209,12 @@ def create_surge_pricing_dag(store_mode: str) -> DAG:
 # instantiate DAGs for Airflow to discover
 dag_airline = create_surge_pricing_dag('airline')
 dag_hotel = create_surge_pricing_dag('hotel')
+
+# TODO: Refactor this factory from a surge pricing factory to a general pricing factory
+# We will do this by passing a pricing strategy class to the factory, since the generic pipeline is:
+# take all interaction data, group by sessionId and assign a new price vector to each session
+# in the grouping we get a subset of the interactions per sessionId and we can map that to some Features
+# we define a custom _get_features(interactions .) methodin the strategy class
+# we then run only the inference which is the .predict(trajectory) per-session which will give us a new price vector
+# this we then publish for each sessionId group
+# this might include no deleting most of the pricers we have defined and starting with a super simple surge-pricing algorithm that is no-fit only predict. This we can then test end-to-end and observe changes to prices according to a desired strategy - we have to define this one as a very short term strategy because we run sessions that take only a few minutes.

experiments/airflow/dags/surge_pricing_pipeline.py

@@ -120,15 +120,31 @@ def apply_surge_pricing(**kwargs):
     # rename demand_score to demand for pricer compatibility
     data = product_features.rename(columns={'demand_score': 'demand'})
 
+    high_thresh = dag_conf.get('high_threshold', 10)
+    low_thresh = dag_conf.get('low_threshold', 2)
+    surge_mult = dag_conf.get('surge_multiplier', 1.2)
+    discount_mult = dag_conf.get('discount_multiplier', 0.9)
+
+    logging.info(f"Surge pricing config: high_thresh={high_thresh}, low_thresh={low_thresh}, surge_mult={surge_mult}, discount_mult={discount_mult}")
+    logging.info(f"Demand stats: min={data['demand'].min():.2f}, max={data['demand'].max():.2f}, mean={data['demand'].mean():.2f}")
+    logging.info(f"Products with high demand (>={high_thresh}): {(data['demand'] >= high_thresh).sum()}")
+    logging.info(f"Products with low demand (<={low_thresh}): {(data['demand'] <= low_thresh).sum()}")
+
     surge_pricer = SimpleSurgePricer(
-        high_threshold=dag_conf.get('high_threshold', 10),
+        high_threshold=high_thresh,
-        low_threshold=dag_conf.get('low_threshold', 2),
+        low_threshold=low_thresh,
-        surge_multiplier=dag_conf.get('surge_multiplier', 1.2),
+        surge_multiplier=surge_mult,
-        discount_multiplier=dag_conf.get('discount_multiplier', 0.9)
+        discount_multiplier=discount_mult
     )
     surge_pricer.fit(data)
     data['optimal_price'] = surge_pricer.predict()
 
+    base_avg = data['base_price'].mean()
+    optimal_avg = data['optimal_price'].mean()
+    price_change_pct = ((optimal_avg - base_avg) / base_avg) * 100
+
+    logging.info(f"Price adjustment: base_avg={base_avg:.2f}, optimal_avg={optimal_avg:.2f}, change={price_change_pct:+.1f}%")
+
     prices_df = data[['productId', 'price', 'base_price', 'optimal_price', 'demand']].rename(columns={
         'price': 'current_price',
         'demand': 'demand_score'

experiments/ml/__init__.py

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

experiments/ml/arch.py

@@ -0,0 +1,122 @@
+# 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

@@ -0,0 +1,103 @@
+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

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

experiments/ml/train.py

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

experiments/procesing/pricers/base.py

@@ -7,15 +7,6 @@ import pandas as pd
 class PricingFunction(ABC):
     """
     Abstract base for pricing functions.
-
-    Defines mapping: f(Q_t, P_t, S_t, H_t) -> P_{t+1}
-
-    Where:
-        Q_t ∈ R^n: demand vector at time t
-        P_t ∈ R^n: price vector at time t
-        S_t: session features (behavioral signals, interactions)
-        H_t = {Q_{t-k}, P_{t-k}, S_{t-k}}: historical state trajectory
-
     Objective:
         maximize E[R_T] = E[Σ P_t^T · Q_t]
         subject to:
@@ -28,10 +19,10 @@ class PricingFunction(ABC):
     def fit(self, *kwargs):
         """
         Offline training on historical data.
+        This is where we can think about some maximization of expected revenue
+        over historical trajectories to learn parameters of the pricing function.
+        (This however we cover move in the RL side of things)
 
-        Args:
-            historical_data: DataFrame with elasticity, prices, demand signals
-            **kwargs: additional training parameters
         """
         pass
 
@@ -39,12 +30,18 @@ class PricingFunction(ABC):
     def predict(self, *kwargs) -> np.ndarray:
         """
         Generate optimal prices given current state.
+        This is an abstract method that transitions from τ -> P*
+        which is the mapping from the trajectory to optimal prices under
+        some subset of session grouping (so, per sessionId)
+        """
+        pass
 
-        Args:
+    @abstractmethod
-            state_space: StateSpace object containing Q_t, P_t, S_t, H_t
+    def _get_features(self, *kwargs) -> np.ndarray:
-
+        """
+        Extract features from trajectory for pricing decision.
         Returns:
-            P_{t+1}: price vector in R^n
+            np.ndarray of shape (n_products, n_features)
         """
         pass
 

experiments/procesing/pricers/elasticity.py

@@ -57,3 +57,13 @@ class ElasticityBasedPricer(PricingFunction):
         # enforce bounds
         prices = np.clip(prices, self.price_floor, self.price_ceil)
         return prices
+
+    def _get_features(self, state_space=None) -> np.ndarray:
+        """Extract elasticity, demand, and demand deviation for each product"""
+        if state_space is None or self.elasticity is None:
+            n = len(self.elasticity) if self.elasticity is not None else 0
+            return np.zeros((n, 3))
+
+        demand = np.asarray(state_space.demand)
+        demand_dev = (demand - self.mean_demand) / (self.mean_demand + 1e-6)
+        return np.column_stack([self.elasticity, demand, demand_dev])

experiments/procesing/pricers/session_aware.py

@@ -107,6 +107,36 @@ class SessionAwarePricer(PricingFunction):
 
         return prices
 
+    def _get_features(self, state_space=None) -> np.ndarray:
+        """Extract elasticity, demand, and session features"""
+        if state_space is None or self.elasticity is None:
+            n = len(self.elasticity) if self.elasticity is not None else 0
+            return np.zeros((n, 5))
+
+        demand = np.asarray(state_space.demand)
+        n_products = len(demand)
+
+        # extract session features
+        velocity = 0.0
+        view_depth = 0.0
+        cart_to_view = 0.0
+
+        if not state_space.session_features.empty:
+            sf = state_space.session_features.iloc[0]
+            velocity = sf.get('interaction_velocity', 0.0)
+            view_depth = sf.get('product_view_depth', 0.0)
+            cart_to_view = sf.get('cart_to_view_ratio', 0.0)
+
+        # broadcast session features to all products
+        features = np.column_stack([
+            self.elasticity,
+            demand,
+            np.full(n_products, velocity),
+            np.full(n_products, view_depth),
+            np.full(n_products, cart_to_view)
+        ])
+        return features
+
 
 class ProductSpecificSessionPricer(PricingFunction):
     """
@@ -170,3 +200,12 @@ class ProductSpecificSessionPricer(PricingFunction):
 
         prices = np.clip(base_prices, self.price_floor, self.price_ceil)
         return prices
+
+    def _get_features(self, state_space=None) -> np.ndarray:
+        """Extract elasticity and demand features for product-specific pricing"""
+        if state_space is None or self.elasticity is None:
+            n = len(self.elasticity) if self.elasticity is not None else 0
+            return np.zeros((n, 2))
+
+        demand = np.asarray(state_space.demand)
+        return np.column_stack([self.elasticity, demand])

experiments/procesing/pricers/simple.py

@@ -3,6 +3,46 @@ import pandas as pd
 from procesing.pricers.base import PricingFunction
 
 
+def session_features_to_demand(session_features: pd.DataFrame) -> float:
+    """
+    Map session behavioral features to demand proxy.
+    THIS is the critical θ̂ → D transformation for rule-based pricing.
+
+    Logic:
+      - High velocity → agent behavior → price up (revenue recovery)
+      - High cart ratio → purchase intent → price up
+      - Low activity → discount to convert
+
+    Returns: demand proxy score (0-20 range, higher = more demand)
+    """
+    if session_features.empty:
+        return 1.0
+
+    feat = session_features.iloc[0] if len(session_features) > 0 else {}
+
+    velocity = feat.get('interaction_velocity', 0)
+    cart_ratio = feat.get('cart_to_view_ratio', 0)
+    item_views = feat.get('item_views', 0)
+    cart_adds = feat.get('cart_adds', 0)
+
+    # baseline demand
+    demand = 1.0
+
+    # agent detection: high velocity → treat as high "demand" to price up
+    if velocity > 2.0:
+        demand += 10.0  # strong agent signal
+
+    # conversion intent: cart interaction → price up
+    if cart_ratio > 0.1 or cart_adds > 0:
+        demand += 5.0
+
+    # browsing depth: many views → interest signal
+    if item_views > 3:
+        demand += min(item_views, 5.0)
+
+    return min(demand, 20.0)  # cap at 20
+
+
 class StaticPricer(PricingFunction):
     """Static pricing: always return fixed base prices"""
 
@@ -25,6 +65,11 @@ class StaticPricer(PricingFunction):
             raise ValueError("Must call fit() or provide base_prices in constructor")
         return self.base_prices.copy()
 
+    def _get_features(self, state_space=None) -> np.ndarray:
+        """Static pricer uses no features, returns empty array"""
+        n = len(self.base_prices) if self.base_prices is not None else 0
+        return np.zeros((n, 0))
+
 
 class RandomPricer(PricingFunction):
     """Random pricing within bounds (for baseline comparison)"""
@@ -47,6 +92,11 @@ class RandomPricer(PricingFunction):
             self.n_products = len(state_space.demand)
         return self.rng.uniform(self.price_min, self.price_max, size=self.n_products)
 
+    def _get_features(self, state_space=None) -> np.ndarray:
+        """Random pricer uses no features"""
+        n = self.n_products if self.n_products else 0
+        return np.zeros((n, 0))
+
 
 class SimpleSurgePricer(PricingFunction):
     """
@@ -70,18 +120,22 @@ class SimpleSurgePricer(PricingFunction):
     def fit(self, market_data: pd.DataFrame):
         """Extract base prices from product catalog or historical averages"""
         self.base_prices = market_data['base_price'].to_numpy() if 'base_price' in market_data.columns else market_data['price'].values
-        self.demand_history = market_data['demand'].to_numpy() if 'demand' in market_data.columns else np.zeros_like(self.base_prices)
+        return self
 
-    def predict(self) -> np.ndarray:
+    def predict(self, state_space) -> np.ndarray:
         """
         Adjust prices based on current demand using surge rules.
-        state_space.demand: demand counts per product
+        state_space.demand: demand proxy per product (from session features)
-        state_space.prices: current prices (fallback if base_prices not set)
+        state_space.prices: base prices
         """
-        current_prices = self.base_prices if self.base_prices is not None else np.ones_like(demand_vector) * 99.99
+        demand = np.asarray(state_space.demand) if state_space and hasattr(state_space, 'demand') else np.array([0])
-        demand = self.demand_history if self.demand_history is not None else np.zeros_like(current_prices)
+        base = np.asarray(state_space.prices) if state_space and hasattr(state_space, 'prices') else self.base_prices
-        new_prices = current_prices.copy()
 
+        if base is None:
+            base = np.ones(len(demand)) * 99.99
+
+        # ensure float dtype to allow multiplication by float multipliers
+        new_prices = base.astype(np.float64).copy()
         high_mask = demand >= self.high_threshold
         new_prices[high_mask] *= self.surge_multiplier
 
@@ -89,3 +143,16 @@ class SimpleSurgePricer(PricingFunction):
         new_prices[low_mask] *= self.discount_multiplier
 
         return new_prices
+
+    def _get_features(self, state_space=None) -> np.ndarray:
+        """Extract demand and base price features for each product"""
+        if state_space is None:
+            n = len(self.base_prices) if self.base_prices is not None else 0
+            return np.zeros((n, 2))
+
+        demand = np.asarray(state_space.demand) if hasattr(state_space, 'demand') else np.array([0])
+        base = np.asarray(state_space.prices) if hasattr(state_space, 'prices') else self.base_prices
+        if base is None:
+            base = np.ones(len(demand)) * 99.99
+
+        return np.column_stack([demand, base])

experiments/procesing/steps/__init__.py

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

experiments/procesing/steps/base.py

@@ -1,6 +1,7 @@
 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):
     """
@@ -16,7 +17,7 @@ class BaseContextStep(BaseEstimator, TransformerMixin, ABC):
         return self
 
     @abstractmethod
-    def transform(self, X):
+    def transform(self, X) -> Any:
         """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: list of price chunks with [productId, price]
+    Output: DataFrame with columns [productId, price]
     """
 
     def transform(self, price_logs_df: pd.DataFrame):
         if price_logs_df.empty:
-            return []
+            return pd.DataFrame(columns=['productId', 'price'])
 
         df = price_logs_df.copy()
         ts_col = self.context.config.get('ts_col', 'ts')

experiments/procesing/steps/session.py

@@ -1,159 +1,262 @@
 """
-Session feature extraction for S_t component of state space.
+Session feature extraction for ML training pipeline.
-Computes behavioral signals from interaction data already in pipeline.
 """
 import pandas as pd
 import numpy as np
-from typing import Optional, Dict, Any
+import re
-from collections import Counter
+from typing import Dict, Any
 from procesing.steps.base import BaseContextStep
 
-def _extract_features_for_session(session_df: pd.DataFrame, session_timeout_sec: float = 900) -> Dict[str, Any]:
+EVENT_CATS = {
-    """Compute features for single session.
+    'page_view': ['page_view'],
-
+    'item_view': ['view_item_page', 'learn_more_about_item'],
-    Args:
+    'cart_add': ['add_item_to_cart'],
-        session_df: interaction events for this session
+    'purchase': ['purchase', 'checkout_complete'],
-        session_timeout_sec: max gap between events before resetting duration (default 900s = 15min)
+    'hover': ['hover_over_title', 'hover_over_paragraph', 'hover_over_link', 'hover_over_button'],
-    """
+    # 'filter': ['filter', 'search', 'apply_filter'],
-    features = {}
+}
-
+HEADLESS_RE = re.compile(r'HeadlessChrome|Headless|PhantomJS', re.I)
-    # basic counts
+AUTOMATION_RE = re.compile(r'Selenium|Playwright|Puppeteer|WebDriver|chromedriver|geckodriver', re.I)
-    features['total_interactions'] = len(session_df)
+BROWSER_PATTERNS = [('Chrome', r'Chrome/[\d.]+'), ('Firefox', r'Firefox/[\d.]+'),
-
+                    ('Safari', r'Safari/[\d.]+'), ('Edge', r'Edg/[\d.]+')]
-    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 _apply_to_slice(df: pd.DataFrame) -> pd.DataFrame:
+def _get_browser(s: str) -> str:
-    """Apply feature extraction to sliding window of interactions."""
+    if pd.isna(s): return 'Unknown'
-    # add columns of all features at each step
+    for name, pat in BROWSER_PATTERNS:
-    new_cols = ["total_interactions", "page_views", "item_views", "searches",
+        if re.search(pat, s): return name
-                "cart_adds", "hovers", "unique_products_viewed", "product_view_depth",
+    return 'Other'
-                "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()
 
 
-        # For simplicity, we return as is
+class TemporalFeatureStep(BaseContextStep):
-        return rich_dataset.copy()
+    """Vectorized time-based features: durations, velocities, gaps."""
+
+    def __init__(self, context, timeout_sec: float = 900, velocity_window: str = '5min'):
+        super().__init__(context)
+        self.timeout_sec = timeout_sec
+        self.velocity_window = velocity_window
+
+    def transform(self, X: pd.DataFrame) -> pd.DataFrame:
+        df = X.copy()
+        if df.empty or 'ts' not in df.columns:
+            return pd.DataFrame(columns=pd.Series(['sessionId']))
+
+        df['ts_dt'] = pd.to_datetime(df['ts'])
+        df = df.sort_values(['sessionId', 'ts_dt'])
+        df['time_diff'] = df.groupby('sessionId')['ts_dt'].diff().dt.total_seconds()
+        df['active_diff'] = df['time_diff'].where(df['time_diff'] <= self.timeout_sec, 0)
+
+        agg = df.groupby('sessionId').agg(
+            session_duration_sec=('active_diff', 'sum'),
+            total_interactions=('sessionId', 'count'),
+            avg_time_between_events=('time_diff', 'mean'),
+            std_time_between_events=('time_diff', 'std'),
+            min_time_between_events=('time_diff', 'min'),
+            session_start_hour=('ts_dt', lambda x: x.min().hour),
+        ).reset_index()
+        agg['std_time_between_events'] = agg['std_time_between_events'].fillna(0)
+        agg['interaction_velocity'] = np.where(
+            agg['session_duration_sec'] > 0,
+            (agg['total_interactions'] / agg['session_duration_sec']) * 60, 0)
+
+        vel = df.set_index('ts_dt').groupby('sessionId').resample(self.velocity_window, include_groups=False).size()
+        max_velocity = vel.groupby('sessionId').max().rename('max_velocity_5min')
+        agg = agg.merge(max_velocity, on='sessionId', how='left')
+        agg['max_velocity_5min'] = agg['max_velocity_5min'].fillna(0)
+        return agg
 
 
+class BehavioralFeatureStep(BaseContextStep):
+    """Vectorized event counts and ratios per session."""
+
+    def transform(self, X: pd.DataFrame) -> pd.DataFrame:
+        df = X.copy()
+        if df.empty or 'eventName' not in df.columns:
+            return pd.DataFrame(columns=pd.Series(['sessionId']))
+
+        for cat, events in EVENT_CATS.items():
+            df[f'is_{cat}'] = df['eventName'].isin(events)
+        df['is_hover'] = df['is_hover'] | df['eventName'].str.startswith('hover_over_')
+
+        agg = df.groupby('sessionId').agg(
+            total_events=('eventName', 'count'), unique_pages=('page', 'nunique'),
+            page_views=('is_page_view', 'sum'), item_views=('is_item_view', 'sum'),
+            cart_adds=('is_cart_add', 'sum'), purchases=('is_purchase', 'sum'),
+            hover_events=('is_hover', 'sum'),
+            # filter_events=('is_filter', 'sum'),
+        ).reset_index()
+        agg['cart_to_view_ratio'] = np.where(agg['item_views'] > 0, agg['cart_adds'] / agg['item_views'], 0)
+        agg['conversion_rate'] = np.where(agg['item_views'] > 0, agg['purchases'] / agg['item_views'], 0)
+        agg['hover_intensity'] = np.where(agg['total_events'] > 0, agg['hover_events'] / agg['total_events'], 0)
+        return agg
+
+
+class ProductFeatureStep(BaseContextStep):
+    """Vectorized product interaction features: diversity, depth, price sensitivity."""
+
+    def transform(self, X: pd.DataFrame) -> pd.DataFrame:
+        df = X.copy()
+        if df.empty:
+            return pd.DataFrame(columns=pd.Series(['sessionId']))
+        price_col = next((c for c in ['metadata_base_price', 'metadata_price', 'base_price'] if c in df.columns), None)
+        df['price_seen'] = pd.to_numeric(df[price_col], errors='coerce') if price_col else np.nan
+
+        prod_df = df[df['productId'].notna()]
+        if prod_df.empty:
+            return pd.DataFrame(columns=pd.Series(['sessionId', 'unique_products_viewed', 'product_view_depth', 'avg_price_seen', 'min_price_seen', 'max_price_seen', 'price_range']))
+
+        agg = prod_df.groupby('sessionId').agg(
+            unique_products_viewed=('productId', 'nunique'),
+            product_view_depth=('productId', lambda x: x.value_counts().iloc[0] if len(x) > 0 else 0),
+            avg_price_seen=('price_seen', 'mean'), min_price_seen=('price_seen', 'min'),
+            max_price_seen=('price_seen', 'max'),
+        ).reset_index()
+        agg['price_range'] = (agg['max_price_seen'] - agg['min_price_seen']).fillna(0)
+        return agg
+
+
+class UserAgentFeatureStep(BaseContextStep):
+    """Parse userAgent into bot-detection signals."""
+
+    def transform(self, X: pd.DataFrame) -> pd.DataFrame|pd.Series:
+        df = X.copy()
+        if df.empty or 'userAgent' not in df.columns:
+            return pd.DataFrame(columns=pd.Series(['sessionId']))
+
+        ua = df.groupby('sessionId')['userAgent'].first().reset_index()
+        ua['is_headless'] = ua['userAgent'].str.contains(HEADLESS_RE, na=False)
+        ua['is_automation'] = ua['userAgent'].str.contains(AUTOMATION_RE, na=False)
+        ua['browser_family'] = ua['userAgent'].apply(_get_browser)
+        return ua[['sessionId', 'is_headless', 'is_automation', 'browser_family']]
 
 
 class ExtractSessionFeaturesStep(BaseContextStep):
     """
-    Extract session-level behavioral features from interaction logs.
+    Vectorized session feature extraction - replaces O(n^2) per-row loop.
-
+    Input: interactions_df
-    Input: interactions_df (user-interactions from earlier pipeline step)
+    Output: session-level feature matrix
-    Output: interactions_df with added session feature columns
+    THIS is our main mapping from tau (trajectory) to some features vector theta - we need to do this very well. This is what will go into demand esimation.
     """
 
-    def transform(self, interactions_df: pd.DataFrame) -> pd.DataFrame:
+    def transform(self, X: pd.DataFrame) -> pd.DataFrame:
-        if interactions_df.empty:
+        if X.empty:
             return pd.DataFrame()
+        df = X.copy()
 
-        # ensure timestamp column
+        # run all feature steps and merge on sessionId
-        if 'ts' in interactions_df.columns:
+        temporal = TemporalFeatureStep(self.context).transform(df)
-            interactions_df = interactions_df.copy()
+        behavioral = BehavioralFeatureStep(self.context).transform(df)
-            interactions_df['ts'] = pd.to_datetime(interactions_df['ts'])
+        product = ProductFeatureStep(self.context).transform(df)
+        ua = UserAgentFeatureStep(self.context).transform(df)
 
-        # group by session and compute features
+        result = temporal
-        session_features = []
+        for other in [behavioral, product, ua]:
-        for session_id, session_df in interactions_df.groupby('sessionId'):
+            if not other.empty and 'sessionId' in other.columns:
-            new_slice = _apply_to_slice(session_df.sort_values('ts'))
+                result = result.merge(other, on='sessionId', how='left')
-            session_features.append(new_slice)
 
-        return pd.concat(session_features, ignore_index=True)
+        # 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
 
 
-
+class JoinLabelsStep(BaseContextStep):
-class FilterSessionInteractionsStep(BaseContextStep):
     """
-    Filter interactions DataFrame to specific session.
+    Join experiment labels to session features.
-
+    Input: (features_df, experiments_df) or features_df (fetches experiments)
-    Input: (interactions_df, session_id)
+    Output: labeled feature matrix with is_agent column
-    Output: interactions_df filtered to session_id
     """
 
-    def transform(self, data: tuple) -> pd.DataFrame:
+    def transform(self, X : tuple) -> pd.DataFrame:
-        interactions_df, session_id = data
+        data = X;
-        return interactions_df[interactions_df['sessionId'] == session_id].copy()
+        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

experiments/procesing/tests/conftest.py

@@ -269,3 +269,13 @@ 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

lib/model_registry.py

@@ -178,3 +178,49 @@ class ModelRegistry:
             return True
         except:
             return False
+
+    def set_session_prices(self, session_id: str, prices: Dict[str, float], ttl: int = 1800):
+        """
+        Store prices for a specific session.
+        THIS is the write path for session-aware pricing.
+
+        Args:
+            session_id: session identifier
+            prices: dict of {productId: price}
+            ttl: time-to-live in seconds (default 30min)
+        """
+        if not prices:
+            return
+
+        key = f"session:{session_id}:prices"
+        # use Redis hash for O(1) lookup per product
+        self.redis_client.hset(key, mapping={k: str(v) for k, v in prices.items()})
+        self.redis_client.expire(key, ttl)
+
+    def get_session_price(self, session_id: str, product_id: str) -> Optional[float]:
+        """
+        Lookup price for (sessionId, productId).
+        THIS is the read path for fast provider lookup.
+
+        Returns: price or None if not found
+        """
+        key = f"session:{session_id}:prices"
+        price_str = self.redis_client.hget(key, product_id)
+
+        if price_str is None:
+            return None
+
+        return float(price_str.decode('utf-8') if isinstance(price_str, bytes) else price_str)
+
+    def get_session_all_prices(self, session_id: str) -> Dict[str, float]:
+        """Get all prices for a session."""
+        key = f"session:{session_id}:prices"
+        prices_raw = self.redis_client.hgetall(key)
+
+        if not prices_raw:
+            return {}
+
+        return {
+            (k.decode('utf-8') if isinstance(k, bytes) else k): float(v.decode('utf-8') if isinstance(v, bytes) else v)
+            for k, v in prices_raw.items()
+        }

sim/rl/behavior_loader/loader.py

@@ -0,0 +1,63 @@
+import os
+from pydantic import BaseModel as Base
+import json
+
+class PayloadModel(Base):
+    sessionId: str
+    experimentId: str | None
+    eventName: str
+    page: str | None
+    productId: str | None
+    metadata: dict
+    storeMode: str
+    userAgent: str
+    ts: str
+
+class ValueModel(Base):
+    payload: PayloadModel
+    encoding: str
+    isPayloadNull: bool
+    schemaId: int
+    size: int
+
+class InteractionModel(Base):
+    partitionID: int
+    offset: int
+    timestamp: int
+    compression: str
+    isTransactional: bool
+    headers: list
+    key: dict
+    value: ValueModel
+
+class Loader:
+    def __init__(self, src_dir: str):
+        self.src_dir = src_dir
+        self.entries = os.listdir(src_dir)
+        if not self.entries: raise ValueError("empty directory")
+        self.data = self._load_sessions()
+
+    def _is_admin_page(self, interaction: InteractionModel) -> bool:
+        page = interaction.value.payload.page
+        return page and page.startswith("/admin/")
+
+    def _load_sessions(self) -> dict:
+        sessions = {}
+        for entry in self.entries:
+            int_path = f"{self.src_dir}/{entry}/int.json"
+            raw = json.load(open(int_path))
+            ints = [InteractionModel(**i) for i in raw]
+            sessions[entry] = [i for i in ints if not self._is_admin_page(i)]
+        return sessions
+
+    def get_data(self) -> dict:
+        return self.data
+
+    def get_entries(self) -> tuple[list[str], int]:
+        return self.entries, len(self.entries)
+
+if __name__ == "__main__":
+    DIR = "/home/velocitatem/Documents/Projects/PHANTOM/experiments/collected_data/"
+    loader = Loader(DIR)
+    _, n = loader.get_entries()
+    print(f"Loaded {n} sessions from {DIR}")

sim/rl/behavior_loader/models.py

@@ -0,0 +1,144 @@
+from loader import Loader
+from collections import defaultdict
+from typing import Dict, List, Tuple, Set
+import numpy as np
+import graphviz
+
+DIR = "/home/velocitatem/Documents/Projects/PHANTOM/experiments/collected_data/"
+
+class BehaviorModel:
+    def __init__(self, src_dir: str = DIR):
+        self.loader = Loader(src_dir)
+        self.data = self.loader.get_data()
+        self.entries, self.num_entries = self.loader.get_entries()
+        self.mdp = None
+
+    def _state_repr(self, evt) -> str:
+        p = evt.value.payload
+        return f"{p.page or 'unk'}|{p.productId or 'none'}|{p.eventName}"
+
+    def _extract_sessions(self):
+        # transform raw events into sequential state trajectories per session
+        trajectories = []
+        for sid, evts in self.data.items():
+            if len(evts) < 2: continue
+            states = [self._state_repr(e) for e in sorted(evts, key=lambda x: x.timestamp)]
+            trajectories.append(states)
+        return trajectories
+
+    def _calc_transitions(self, trajectories: List[List[str]]) -> Tuple[Dict, Set]:
+        trans = defaultdict(lambda: defaultdict(int))
+        states = set()
+        for traj in trajectories:
+            for i in range(len(traj) - 1):
+                s, s_next = traj[i], traj[i+1]
+                trans[s][s_next] += 1
+                states.update([s, s_next])
+        return trans, states
+
+    def _calc_rewards(self, trajectories: List[List[str]]) -> Dict:
+        # reward based on session progression depth
+        rwd = defaultdict(list)
+        for traj in trajectories:
+            n = len(traj)
+            for i, s in enumerate(traj):
+                rwd[s].append(i / n)
+        return rwd
+
+    def _normalize_trans(self, counts: Dict) -> Dict:
+        return {s: {s_n: cnt/sum(nxt.values()) for s_n, cnt in nxt.items()}
+                for s, nxt in counts.items()}
+
+    def build_MDP(self) -> Dict:
+        trajs = self._extract_sessions()
+        trans_cnt, states = self._calc_transitions(trajs)
+        trans_prob = self._normalize_trans(trans_cnt)
+        state_rwd = self._calc_rewards(trajs)
+        state_val = {s: np.mean(r) for s, r in state_rwd.items()}
+
+        self.mdp = {
+            'states': sorted(list(states)),
+            'num_states': len(states),
+            'transitions': trans_prob,
+            'state_values': state_val,
+            'state_rewards': state_rwd,
+            'trans_counts': trans_cnt,
+        }
+        return self.mdp
+
+    def transition_prob(self, s: str, s_next: str) -> float:
+        if not self.mdp: raise ValueError("build MDP first")
+        return self.mdp['transitions'].get(s, {}).get(s_next, 0.0)
+
+    def state_value(self, s: str) -> float:
+        if not self.mdp: raise ValueError("build MDP first")
+        return self.mdp['state_values'].get(s, 0.0)
+
+    def sample_traj(self, start: str, max_len: int = 50) -> List[str]:
+        if not self.mdp: raise ValueError("build MDP first")
+        path = [start]
+        curr = start
+        for _ in range(max_len):
+            nxt = self.mdp['transitions'].get(curr, {})
+            if not nxt: break
+            curr = np.random.choice(list(nxt.keys()), p=list(nxt.values()))
+            path.append(curr)
+        return path
+
+def visualize_mdp(model: BehaviorModel, threshold: float = 0.05, output: str = "mdp_graph", fmt: str = "svg", view: bool = False, export_dot: bool = False):
+    """visualize MDP as directed graph using graphviz, aggregated by event type"""
+    if not model.mdp: raise ValueError("build MDP first")
+
+    # aggregate transitions by event type
+    evt_trans = defaultdict(lambda: defaultdict(float))
+    for s, trans in model.mdp['transitions'].items():
+        evt_src = s.split('|')[2]
+        for s_next, prob in trans.items():
+            evt_dst = s_next.split('|')[2]
+            evt_trans[evt_src][evt_dst] += prob
+
+    # normalize aggregated transitions
+    for evt_src in evt_trans:
+        total = sum(evt_trans[evt_src].values())
+        if total > 0:
+            for evt_dst in evt_trans[evt_src]:
+                evt_trans[evt_src][evt_dst] /= total
+
+    g = graphviz.Digraph(format=fmt)
+    g.attr(rankdir='LR', size='30')
+    g.attr('node', shape='circle', width='1', height='1')
+
+    # collect all event types
+    events = set(evt_trans.keys())
+    for trans in evt_trans.values():
+        events.update(trans.keys())
+
+    # add nodes for each event type
+    for evt in events:
+        g.node(evt)
+
+    # add edges above threshold
+    for evt_src in evt_trans:
+        for evt_dst, prob in evt_trans[evt_src].items():
+            if prob > threshold:
+                g.edge(evt_src, evt_dst, label=f'{prob:.2f}')
+
+    g.render(output, view=view, cleanup=True)
+    print(f"Saved MDP graph to {output}.{fmt}")
+
+    if export_dot:
+        dot_file = f"{output}.dot"
+        with open(dot_file, 'w') as f:
+            f.write(g.source)
+        print(f"Exported DOT source to {dot_file}")
+
+    return g
+
+if __name__ == "__main__":
+    model = BehaviorModel(DIR)
+    mdp = model.build_MDP()
+    print(f"Built MDP: {mdp['num_states']} states, {sum(len(t) for t in mdp['transitions'].values())} transitions")
+    if not mdp['states']:
+        print("No states found")
+        exit(1)
+    visualize_mdp(model, threshold=0.05, output="mdp_viz", fmt="pdf", export_dot=True)

sim/rl/engine.py

@@ -0,0 +1,227 @@
+from os import kill
+import numpy as np
+import pandas as pd
+from abc import ABC, abstractmethod
+from typing import Dict, Any
+from environment import BusinessLogicConstraints
+
+
+"""
+An angine by default should have its own demand estimation mechanism from the observed observations whihc are the computer feature.
+From these features we then follow the researc hstructure of q -> p with a testable and must be updatable mechanism.
+"""
+
+class BasePricingEngine(ABC):
+    """base interface for all pricing engines"""
+    def __init__(self, constraints: BusinessLogicConstraints, seed: int = 0):
+        self.c = constraints
+        self.rng = np.random.default_rng(seed)
+        self.step_count = 0
+
+
+    @abstractmethod
+    def compute_prices(self, current_prices: np.ndarray, observation: Dict[str, Any]) -> np.ndarray:
+        """compute new prices given current state and observation from environment
+
+        args:
+            current_prices: current price vector [N]
+            observation: dict containing 'price', 'demand', and possibly interaction data
+
+        returns:
+            new_prices: updated price vector [N]
+        """
+        pass
+
+    @abstractmethod
+    def update(obs, reward, done, info):
+        pass
+
+
+
+    def reset(self):
+        """reset engine state for new episode"""
+        self.step_count = 0
+
+
+class WildPricingEngine(BasePricingEngine):
+    """production-like pricing using online elasticity estimation via EWMA regression"""
+    def __init__(self, constraints: BusinessLogicConstraints, seed: int = 0):
+        super().__init__(constraints, seed)
+        # per-product unit costs (unknown to customers; known to platform)
+        self.unit_cost = self.rng.uniform(8.0, 40.0, size=self.c.product_catelogue_size).astype(np.float32)
+        # online elasticity estimate (start moderately elastic)
+        self.e_hat = np.full((self.c.product_catelogue_size,), -1.3, dtype=np.float32)
+        # EWMA state for log-log regression
+        self.mu_logp = np.zeros(self.c.product_catelogue_size, dtype=np.float32)
+        self.mu_logq = np.zeros(self.c.product_catelogue_size, dtype=np.float32)
+        self.cov_pq  = np.zeros(self.c.product_catelogue_size, dtype=np.float32)
+        self.var_p   = np.ones(self.c.product_catelogue_size, dtype=np.float32)
+        # knobs typical in production
+        self.lr = 0.08
+        self.ewma = 0.05
+        self.eps_explore = 0.03
+        self.explore_scale = 0.03
+
+    def _safe_elasticity(self, e: np.ndarray) -> np.ndarray:
+        return np.clip(e, -5.0, -1.05)
+
+    def reset(self):
+        super().reset()
+        self.e_hat = np.full((self.c.product_catelogue_size,), -1.3, dtype=np.float32)
+        self.mu_logp = np.zeros(self.c.product_catelogue_size, dtype=np.float32)
+        self.mu_logq = np.zeros(self.c.product_catelogue_size, dtype=np.float32)
+        self.cov_pq = np.zeros(self.c.product_catelogue_size, dtype=np.float32)
+        self.var_p = np.ones(self.c.product_catelogue_size, dtype=np.float32)
+
+    def compute_prices(self, current_prices: np.ndarray, observation: Dict[str, Any]) -> np.ndarray:
+        self.step_count += 1
+        # extract demand signal (from env observation) as proxy for sales
+        demand = observation.get('demand', np.zeros(self.c.product_catelogue_size, dtype=np.float32))
+        return self._update_from_demand(current_prices, demand)
+
+    def _update_from_demand(self, prices: np.ndarray, sold: np.ndarray) -> np.ndarray:
+        # log transforms (add 1 to handle zeros)
+        logp = np.log(np.clip(prices, 1e-3, None)).astype(np.float32)
+        logq = np.log(sold + 1.0).astype(np.float32)
+        # EWMA moments for per-product regression: logq ≈ a + e*logp
+        a = self.ewma
+        dp = logp - self.mu_logp
+        dq = logq - self.mu_logq
+        self.mu_logp = (1 - a) * self.mu_logp + a * logp
+        self.mu_logq = (1 - a) * self.mu_logq + a * logq
+        self.cov_pq = (1 - a) * self.cov_pq + a * (dp * dq)
+        self.var_p = (1 - a) * self.var_p + a * (dp * dp + 1e-6)
+        e_new = self.cov_pq / (self.var_p + 1e-6)
+        self.e_hat = self._safe_elasticity(0.9 * self.e_hat + 0.1 * e_new)
+        # profit-optimal price for isoelastic demand (if e < -1)
+        e = self.e_hat
+        p_star = self.unit_cost * (e / (e + 1.0))
+        # smooth toward p_star
+        new_prices = (1 - self.lr) * prices + self.lr * p_star
+        # exploration (small random perturbations)
+        if self.rng.random() < self.eps_explore:
+            noise = self.rng.normal(0.0, self.explore_scale, size=new_prices.shape).astype(np.float32)
+            new_prices = new_prices * (1.0 + noise)
+        # apply business guardrails (max change + bounds)
+        max_adj = self.c.max_price_adjustment
+        ratio = np.clip(new_prices / (prices + 1e-6), 1 - max_adj, 1 + max_adj)
+        new_prices = prices * ratio
+        new_prices = np.clip(new_prices, self.c.system_min_price, self.c.system_max_price).astype(np.float32)
+        return new_prices
+
+
+class StaticPricingEngine(BasePricingEngine):
+    """baseline: fixed prices throughout episode"""
+    def __init__(self, constraints: BusinessLogicConstraints, seed: int = 0):
+        super().__init__(constraints, seed)
+        self.fixed_prices = None
+
+    def reset(self):
+        super().reset()
+        self.fixed_prices = None
+
+    def compute_prices(self, current_prices: np.ndarray, observation: Dict[str, Any]) -> np.ndarray:
+        self.step_count += 1
+        if self.fixed_prices is None:
+            self.fixed_prices = current_prices.copy()
+        return self.fixed_prices.copy()
+
+
+class SimpleDemandEngine(BasePricingEngine):
+    """demand-driven pricing: increase price when demand rises, decrease when it falls"""
+    def __init__(self, constraints: BusinessLogicConstraints, seed: int = 0):
+        super().__init__(constraints, seed)
+        self.prev_demand = None
+        self.lr = 0.05
+
+    def reset(self):
+        super().reset()
+        self.prev_demand = None
+
+    def compute_prices(self, current_prices: np.ndarray, observation: Dict[str, Any]) -> np.ndarray:
+        self.step_count += 1
+        demand = observation.get('demand', np.zeros(self.c.product_catelogue_size, dtype=np.float32))
+        if self.prev_demand is None:
+            self.prev_demand = demand.copy()
+            return current_prices.copy()
+        # simple rule: if demand increases, raise price; if decreases, lower price
+        delta_d = demand - self.prev_demand
+        price_adj = self.lr * np.sign(delta_d) * np.abs(delta_d) / (np.abs(self.prev_demand) + 1.0)
+        new_prices = current_prices * (1.0 + price_adj)
+        self.prev_demand = demand.copy()
+        # apply constraints
+        max_adj = self.c.max_price_adjustment
+        ratio = np.clip(new_prices / (current_prices + 1e-6), 1 - max_adj, 1 + max_adj)
+        new_prices = current_prices * ratio
+        return np.clip(new_prices, self.c.system_min_price, self.c.system_max_price).astype(np.float32)
+
+
+class RandomWalkEngine(BasePricingEngine):
+    """random walk pricing with mean reversion"""
+    def __init__(self, constraints: BusinessLogicConstraints, seed: int = 0):
+        super().__init__(constraints, seed)
+        self.target_price = None
+        self.volatility = 0.02
+
+    def reset(self):
+        super().reset()
+        self.target_price = None
+
+    def compute_prices(self, current_prices: np.ndarray, observation: Dict[str, Any]) -> np.ndarray:
+        self.step_count += 1
+        if self.target_price is None:
+            self.target_price = current_prices.copy()
+        # random walk with mean reversion toward target
+        noise = self.rng.normal(0.0, self.volatility, size=current_prices.shape).astype(np.float32)
+        reversion = 0.01 * (self.target_price - current_prices)
+        new_prices = current_prices * (1.0 + noise) + reversion
+        # apply constraints
+        max_adj = self.c.max_price_adjustment
+        ratio = np.clip(new_prices / (current_prices + 1e-6), 1 - max_adj, 1 + max_adj)
+        new_prices = current_prices * ratio
+        return np.clip(new_prices, self.c.system_min_price, self.c.system_max_price).astype(np.float32)
+
+
+class ThompsonSamplingEngine(BasePricingEngine):
+    """bayesian bandit approach per product treating price as discrete action"""
+    def __init__(self, constraints: BusinessLogicConstraints, seed: int = 0):
+        super().__init__(constraints, seed)
+        self.n_price_levels = 5
+        self.alpha = np.ones((self.c.product_catelogue_size, self.n_price_levels), dtype=np.float32)
+        self.beta = np.ones((self.c.product_catelogue_size, self.n_price_levels), dtype=np.float32)
+        self.price_grid = None
+        self.last_actions = None
+
+    def reset(self):
+        super().reset()
+        self.alpha = np.ones((self.c.product_catelogue_size, self.n_price_levels), dtype=np.float32)
+        self.beta = np.ones((self.c.product_catelogue_size, self.n_price_levels), dtype=np.float32)
+        self.price_grid = None
+        self.last_actions = None
+
+    def compute_prices(self, current_prices: np.ndarray, observation: Dict[str, Any]) -> np.ndarray:
+        self.step_count += 1
+        if self.price_grid is None:
+            # define price grid per product
+            lo = current_prices * 0.7
+            hi = current_prices * 1.3
+            self.price_grid = np.linspace(lo, hi, self.n_price_levels).T
+        demand = observation.get('demand', np.zeros(self.c.product_catelogue_size, dtype=np.float32))
+        # update beliefs based on last action
+        if self.last_actions is not None:
+            for i in range(self.c.product_catelogue_size):
+                a = self.last_actions[i]
+                reward = demand[i]
+                if reward > 0.5:
+                    self.alpha[i, a] += reward
+                else:
+                    self.beta[i, a] += 1.0
+        # thompson sampling: sample from posterior, pick best
+        new_prices = np.zeros(self.c.product_catelogue_size, dtype=np.float32)
+        actions = np.zeros(self.c.product_catelogue_size, dtype=int)
+        for i in range(self.c.product_catelogue_size):
+            theta = self.rng.beta(self.alpha[i], self.beta[i]).astype(np.float32)
+            actions[i] = int(np.argmax(theta))
+            new_prices[i] = self.price_grid[i, actions[i]]
+        self.last_actions = actions
+        return np.clip(new_prices, self.c.system_min_price, self.c.system_max_price).astype(np.float32)

sim/rl/environment.py

@@ -0,0 +1,320 @@
+from sys import intern
+import gymnasium as gym
+from gymnasium import spaces
+from matplotlib import interactive
+import numpy as np
+from dataclasses import dataclass
+import pandas as pd
+from typing import Callable, Optional, Dict, Any, List
+
+# "learner"  agent learning to optimize pricing
+# "agent"  part of environment creating demand signals that learner processes
+
+@dataclass
+class BusinessLogicConstraints():
+    max_price_adjustment: float = 0.30
+    system_max_price: float = 500.0
+    system_min_price: float = 1.0
+    product_catelogue_size: int = 100
+    episode_length: int = 200
+    sessions_per_step: int = 250
+    agent_share: float = 0.25
+    agent_recon_multiplier: float = 6.0
+    agent_purchase_probability: float = 0.20
+    coi_strength: float = 0.25
+    coi_threshold: float = 4.0
+    coi_sigmoid_temp: float = 1.25
+    base_human_demand: float = 0.08
+    base_agent_demand: float = 0.05
+    human_price_elasticity: float = -1.2 # assumptions here
+    agent_price_elasticity: float = -0.6
+    w_agent_loss: float = 1.0
+    w_volatility: float = 5.0
+    w_estimation_error: float = 0.25
+    seed: int = 7
+
+
+def _sigmoid(x: np.ndarray) -> np.ndarray:
+    return 1.0 / (1.0 + np.exp(-x))
+
+class CommercePlatform:
+    """
+    This is just an extension of the state management for the environment, it does not implement anything dynamic just helps us simulate demand.
+    """
+    def __init__(self,
+                 product_catelogue_size: int,
+                 max_price: float,
+                 min_price: float,
+                 constraints: BusinessLogicConstraints):
+        self.product_catelogue_size = product_catelogue_size
+        self.product_supply = np.random.uniform(low=10, high=50, size=(self.product_catelogue_size,))
+        self.max_price = max_price
+        self.min_price = min_price
+        self.constraints = constraints
+        self.simulation_history: List[Dict[str, Any]] = []
+        self._rng = np.random.default_rng(constraints.seed)
+        self._last_interaction_df: pd.DataFrame = pd.DataFrame()
+
+
+    def setup_true_demand(self, prices: np.ndarray) -> Dict[str, np.ndarray]:
+        # ground truth purchase propensities
+        p = np.clip(prices, self.min_price, self.max_price)
+        pn = p / self.max_price
+        human_prob = self.constraints.base_human_demand * (pn ** self.constraints.human_price_elasticity)
+        agent_prob = self.constraints.base_agent_demand * (pn ** self.constraints.agent_price_elasticity)
+        return {
+            "human_purchase_prob": np.clip(human_prob, 0.0, 0.95),
+            "agent_purchase_prob": np.clip(agent_prob, 0.0, 0.95)
+        }
+
+    def _load_behavioral_profile(actor : str, demand_forcing):
+        """
+        This returns a markov chain with average weights which we get from interaction data of our experiments.
+        This defines transition probabilities between different events:
+        search -> view_item_price_binN: 0.7
+        view_item_price_binN -> add_to_cart: 0.2
+        we also must reweight with the demand_forcing vector or purchase probabilities per-product
+        """
+
+
+    def _simulate_sessions(self, base_prices: np.ndarray) -> pd.DataFrame:
+        demand = self.setup_true_demand(base_prices)
+        human_pprob = demand["human_purchase_prob"]
+        agent_pprob = demand["agent_purchase_prob"]
+        events: List[Dict[str, Any]] = []
+        T = self.constraints.sessions_per_step
+        n_agent_sessions = int(round(T * self.constraints.agent_share))
+        n_human_sessions = T - n_agent_sessions
+        n_agent_ids = max(1, n_agent_sessions // 2)
+        session_map = {
+            'humans': n_human_sessions,
+            'agents': n_agent_ids
+        }
+        pprob_map = {
+            'humans': human_pprob,
+            'agents': agent_pprob
+        }
+        joint_events = []
+        for actor, n_sessions in session_map.items():
+            bp = _load_behavioral_profile(actor, pprob_map[actor])
+            counter = 0
+            events = []
+            while counter < n_sessions:
+                session_events = []
+                while len(session_events) == 0 or session_events[-1]['action'] == 'checkout':
+                    interaction_event = bp.sample(self._rng)
+                    interaction_event['session_id'] = f'{actor}_{counter:06d}'
+                    # TODO any other assignments
+                    session_events.append(interaction_event)
+                events.extend(session_events)
+                counter += 1
+            joint_events.extend(events)
+
+        return pd.DataFrame(joint_events)
+
+    def compute_interaction_features(self, interaction_df: pd.DataFrame) -> Dict[str, float]:
+        if interaction_df.empty:
+            return {"mean_sale_price": 0.0, "look_to_book": 0.0}
+        purchases = interaction_df[interaction_df["action"] == "purchase"]
+        mean_sale_price = float(purchases["price_paid"].mean()) if not purchases.empty else 0.0
+        views = float((interaction_df["action"] == "view").sum())
+        buys = float((interaction_df["action"] == "purchase").sum())
+        return {"mean_sale_price": mean_sale_price, "look_to_book": float(views / (buys + 1e-6))}
+
+    def _session_feature_table(self, df: pd.DataFrame) -> pd.DataFrame:
+        # TODO: adapt this
+        if df.empty:
+            return pd.DataFrame()
+        g = df.groupby("session_id", sort=False)
+        session_duration = g["t"].max() - g["t"].min()
+        total_interactions = g.size()
+        avg_time_between = g["t"].apply(lambda x: float(np.diff(np.sort(x.to_numpy())).mean()) if len(x) > 1 else 0.0)
+        interaction_velocity = total_interactions / (session_duration + 1e-6)
+        views = g.apply(lambda x: int((x["action"] == "view").sum()), include_groups=False)
+        cart_adds = g.apply(lambda x: int((x["action"] == "cart").sum()), include_groups=False)
+        purchases = g.apply(lambda x: int((x["action"] == "purchase").sum()), include_groups=False)
+        conversion_rate = purchases / (views + 1e-6)
+        is_agent = g["actor"].apply(lambda s: bool((s == "agent").any()), include_groups=False)
+
+        return pd.DataFrame({
+            "session_duration_sec": session_duration.astype(float),
+            "avg_time_between_events": avg_time_between.astype(float),
+            "total_interactions": total_interactions.astype(int),
+            "interaction_velocity": interaction_velocity.astype(float),
+            "item_views": views.astype(int),
+            "cart_adds": cart_adds.astype(int),
+            "purchases": purchases.astype(int),
+            "conversion_rate": conversion_rate.astype(float),
+            "is_agent": is_agent.astype(bool),
+        }).reset_index()
+
+    def get_interaction_data(self) -> np.ndarray:
+        if self._last_interaction_df.empty:
+            return np.array([], dtype=object)
+        return self._last_interaction_df.to_dict(orient="records")
+
+
+class PHANTOMEnv(gym.Env):
+    metadata = {"render_modes": []}
+
+    def __init__(self, constraints):
+        super().__init__()
+        self.constraints = BusinessLogicConstraints()
+        self.action_space = spaces.Box(low=-self.constraints.max_price_adjustment,
+                                       high=self.constraints.max_price_adjustment,
+                                       shape=(self.constraints.product_catelogue_size,), dtype=np.float32)
+        self.observation_space = spaces.Dict({
+            "elasticity": spaces.Dict({
+                "price": spaces.Box(
+                    low=np.full((self.constraints.product_catelogue_size,), self.constraints.system_min_price, dtype=np.float32),
+                    high=np.full((self.constraints.product_catelogue_size,), self.constraints.system_max_price, dtype=np.float32),
+                    dtype=np.float32),
+                "demand": spaces.Box(
+                    low=np.zeros((self.constraints.product_catelogue_size,), dtype=np.float32),
+                    high=np.full((self.constraints.product_catelogue_size,), 1e6, dtype=np.float32),
+                    dtype=np.float32),
+            })
+            # TODO: define more features that we compute from the interaction data
+        })
+        self.commerce_platform = CommercePlatform(
+            product_catelogue_size=self.constraints.product_catelogue_size,
+            max_price=self.constraints.system_max_price,
+            min_price=self.constraints.system_min_price,
+            constraints=self.constraints)
+        self._rng = np.random.default_rng(self.constraints.seed)
+        self.t = 0
+        self._prev_prices: Optional[np.ndarray] = None
+        self.state: Dict[str, Any] = {}
+
+    def reset(self, seed: Optional[int] = None, options: Optional[dict] = None):
+        super().reset(seed=seed)
+        if seed is not None:
+            self._rng = np.random.default_rng(seed)
+            self.commerce_platform._rng = np.random.default_rng(seed)
+        self.t = 0
+        init_prices = self._rng.uniform(low=60.0, high=140.0, size=(self.constraints.product_catelogue_size,)).astype(np.float32)
+        self._prev_prices = init_prices.copy()
+        self.state = {
+            "elasticity": {
+                "price": init_prices,
+                "demand": np.zeros((self.constraints.product_catelogue_size,), dtype=np.float32),
+            }
+        }
+        return self.state, {}
+
+    def step(self, action: np.ndarray):
+        self.t += 1
+        base_prices = self.state["elasticity"]["price"].astype(np.float32)
+        new_prices = np.clip(base_prices * (1.0 + action.astype(np.float32)),
+                           self.constraints.system_min_price,
+                           self.constraints.system_max_price).astype(np.float32)
+
+        self.state["elasticity"]["price"] = new_prices
+        # TODO: use the commerce platform to simulate sessions
+        interactions_df = self.commerce_platform._simulate_sessions(new_prices)
+        result = self.commerce_platform.compute_interaction_features(interactions_df)
+        # TODO: implement COI computation to use in reward
+        COI = 0.0
+
+        volatility = 0.0 if self._prev_prices is None else \
+            float(np.mean(np.abs((new_prices - self._prev_prices) / (self._prev_prices + 1e-6))))
+        self._prev_prices = new_prices.copy()
+
+        revenue_observed = float(result["revenue_observed"])
+        agent_loss = float(result["agent_loss"])
+
+        reward = (revenue_observed
+                  - COI
+                  - self.constraints.w_agent_loss * agent_loss
+                  - self.constraints.w_volatility * volatility
+                  - self.constraints.w_estimation_error
+                  )
+
+        terminated = self.t >= self.constraints.episode_length
+        info = {
+            "t": self.t,
+            "revenue_observed": revenue_observed,
+            "revenue_oracle": float(result["revenue_oracle"]),
+            "agent_loss": agent_loss,
+            "ux_volatility": volatility,
+            "mean_internal_error": err_mean,
+            "look_to_book": float(result["interaction_features"].get("look_to_book", 0.0)),
+            "mean_sale_price": float(result["interaction_features"].get("mean_sale_price", 0.0)),
+            "true_human_purchases_total": float(np.sum(result["true_human_demand"])),
+            "true_agent_purchases_total": float(np.sum(result["true_agent_purchases"])),
+        }
+        return self.state, float(reward), terminated, False, info
+
+
+if __name__ == "__main__":
+    import matplotlib.pyplot as plt
+    from collections import defaultdict
+
+    runs = {}
+    for use_defense in (False, True):
+        env = PHANTOMEnv(use_defense=use_defense)
+        obs, _ = env.reset(seed=42)
+        metrics = defaultdict(list)
+        total_reward = 0.0
+        done = False
+
+        while not done:
+            action = env.action_space.sample()
+            obs, reward, done, _, info = env.step(action)
+            total_reward += reward
+            p_mean = float(np.mean(obs["elasticity"]["price"]))
+            q_mean = float(np.mean(obs["elasticity"]["demand"]))
+            p_std = float(np.std(obs["elasticity"]["price"]))
+
+            metrics['t'].append(info['t'])
+            metrics['price_mean'].append(p_mean)
+            metrics['price_std'].append(p_std)
+            metrics['demand_mean'].append(q_mean)
+            metrics['revenue_observed'].append(info['revenue_observed'])
+            metrics['revenue_oracle'].append(info['revenue_oracle'])
+            metrics['agent_loss'].append(info['agent_loss'])
+            metrics['ux_volatility'].append(info['ux_volatility'])
+            metrics['look_to_book'].append(info['look_to_book'])
+            metrics['reward'].append(reward)
+            metrics['human_purchases'].append(info['true_human_purchases_total'])
+            metrics['agent_purchases'].append(info['true_agent_purchases_total'])
+
+            if info['t'] % 20 == 0 or done:
+                print(f"defense={'ON ' if use_defense else 'OFF'} t={info['t']:03d} p={p_mean:6.2f}±{p_std:4.2f} "
+                      f"q={q_mean:6.2f} rev={info['revenue_observed']:7.2f} oracle={info['revenue_oracle']:7.2f} "
+                      f"loss={info['agent_loss']:6.2f} ux={info['ux_volatility']:.3f} "
+                      f"ltb={info['look_to_book']:5.2f} r={reward:7.2f}")
+
+        runs[use_defense] = metrics
+        print(f"defense={'ON ' if use_defense else 'OFF'} total_reward={total_reward:.2f}\n")
+
+    fig, axes = plt.subplots(3, 3, figsize=(15, 12))
+    fig.suptitle('PHANTOM Environment: Defense OFF vs ON', fontsize=14, fontweight='bold')
+
+    plot_configs = [
+        ('price_mean', 'Mean Price', 'Price'),
+        ('demand_mean', 'Mean Demand Estimate', 'Demand'),
+        ('revenue_observed', 'Revenue (Observed)', 'Revenue'),
+        ('agent_loss', 'Agent Loss (Oracle - Observed)', 'Loss'),
+        ('ux_volatility', 'UX Volatility (Price Change)', 'Volatility'),
+        ('look_to_book', 'Look-to-Book Ratio', 'Ratio'),
+        ('reward', 'Step Reward', 'Reward'),
+        ('human_purchases', 'Human Purchases', 'Count'),
+        ('agent_purchases', 'Agent Purchases', 'Count'),
+    ]
+
+    for idx, (key, title, ylabel) in enumerate(plot_configs):
+        ax = axes[idx // 3, idx % 3]
+        for use_defense, label, color in [(False, 'No Defense', 'red'), (True, 'With Defense', 'blue')]:
+            m = runs[use_defense]
+            ax.plot(m['t'], m[key], label=label, color=color, alpha=0.7, linewidth=1.5)
+        ax.set_xlabel('Step')
+        ax.set_ylabel(ylabel)
+        ax.set_title(title, fontsize=10, fontweight='bold')
+        ax.legend(loc='best', fontsize=8)
+        ax.grid(True, alpha=0.3)
+
+    plt.tight_layout()
+    plt.savefig('phantom_env_comparison.png', dpi=150, bbox_inches='tight')
+    print("Plot saved to phantom_env_comparison.png")
+    plt.show()

sim/rl/train.py

@@ -0,0 +1,149 @@
+import numpy as np
+import logging
+from pathlib import Path
+from typing import Dict, Type, Optional
+import pickle
+from torch import neg_
+from torch.utils.tensorboard import SummaryWriter
+from environment import PHANTOMEnv, FastTrainingConstraints, BusinessLogicConstraints
+from engine import (BasePricingEngine, WildPricingEngine, StaticPricingEngine,
+                   SimpleDemandEngine, RandomWalkEngine, ThompsonSamplingEngine)
+
+logging.basicConfig(level=logging.INFO, format='%(asctime)s [%(levelname)s] %(message)s')
+logger = logging.getLogger(__name__)
+
+
+
+"""
+Target training loop:
+have base prices p0 from env reset and run the env step, collect reward and metrics
+pass this to the pricing engine which computes the price action to take based on previous reward by learning
+the new action gets passed to the step
+so we alternate, step -> reward -> engine (produces price delta) -> step with price delta -> reward
+to make sure the reinforcement learning inside the engine can learn we need to have trajectory of prices
+CURRENT SOLUTION BELOW does not implement correct learning or updates.
+"""
+
+class EngineTrainer:
+    """wrapper to run pricing engines through episodes and collect metrics"""
+    def __init__(self, engine: BasePricingEngine, env: PHANTOMEnv,
+                 tb_writer: Optional[SummaryWriter] = None):
+        self.engine = engine
+        self.env = env
+        self.episode_metrics = []
+        self.tb_writer = tb_writer
+        self.global_step = 0
+
+    def train(self, n_episodes: int, seed: int = 42):
+
+        obs, _ = self.env.reset(seed=seed)
+        prices = None
+        for ep in range(n_episodes):
+            prices = self.engine.compute_prices(prices, obs)
+            obs, reward, done, _, info = self.env.step(prices)
+            self.engine.update(obs, reward, done, info)
+        return self
+
+
+
+
+
+
+        return self.episode_metrics
+
+    def evaluate(self, n_episodes: int = 10, seed: int = 100) -> Dict:
+        """evaluate trained engine"""
+        results = {k: [] for k in ['total_reward', 'revenue_observed', 'revenue_oracle',
+                                   'agent_loss', 'ux_volatility', 'look_to_book']}
+        for ep in range(n_episodes):
+            metrics = self.run_episode(seed=seed + ep)
+            for k in results:                results[k].append(metrics[k])
+        return {k: (np.mean(v), np.std(v)) for k, v in results.items()}
+
+
+def make_env(fast: bool = True):
+    constraints = FastTrainingConstraints() if fast else BusinessLogicConstraints()
+    return PHANTOMEnv(constraints=constraints)
+
+
+def train_engine(engine_cls: Type[BasePricingEngine], env: PHANTOMEnv,
+                n_episodes: int, seed: int = 42,
+                tb_writer: Optional[SummaryWriter] = None) -> EngineTrainer:
+    constraints = env.constraints
+    engine = engine_cls(constraints=constraints, seed=seed)
+    trainer = EngineTrainer(engine, env, tb_writer=tb_writer)
+    trainer.train(n_episodes, seed=seed)
+    return trainer
+
+
+def save_trainer(trainer: EngineTrainer, path: Path):
+    """save engine state and metrics"""
+    path.parent.mkdir(parents=True, exist_ok=True)
+    with open(path, 'wb') as f:
+        pickle.dump({
+            'engine': trainer.engine,
+            'metrics': trainer.episode_metrics
+        }, f)
+    logger.info(f"Saved trainer to {path}")
+
+
+def load_trainer(path: Path, env: PHANTOMEnv,
+                 tb_writer: Optional[SummaryWriter] = None) -> EngineTrainer:
+    """load saved engine"""
+    with open(path, 'rb') as f:
+        data = pickle.load(f)
+    trainer = EngineTrainer(data['engine'], env, tb_writer=tb_writer)
+    trainer.episode_metrics = data['metrics']
+    return trainer
+
+
+if __name__ == "__main__":
+    base_dir = Path("./runs")
+    base_dir.mkdir(exist_ok=True)
+
+    engines = {
+        "Wild": WildPricingEngine,
+        "Static": StaticPricingEngine,
+#        "SimpleDemand": SimpleDemandEngine,
+        "RandomWalk": RandomWalkEngine,
+        "ThompsonSampling": ThompsonSamplingEngine,
+    }
+    defenses = [False, True]
+    n_train_episodes = 50
+    n_eval_episodes = 10
+    seed = 42
+    fast_mode = True
+
+    logger.info(f"Training config: {n_train_episodes} episodes per engine, fast_mode={fast_mode}")
+
+    trained_trainers = {}
+
+    for engine_name, engine_cls in engines.items():
+        for use_defense in defenses:
+            defense_label = "defense_on" if use_defense else "defense_off"
+            run_name = f"{engine_name}_{defense_label}"
+            log_dir = base_dir / run_name
+            log_dir.mkdir(parents=True, exist_ok=True)
+
+            logger.info(f"Training {engine_name} with defense={use_defense}")
+            logger.info(f"Log directory: {log_dir}")
+
+            env = make_env(fast=fast_mode)
+            tb_writer = SummaryWriter(log_dir=str(log_dir))
+            trainer = train_engine(engine_cls, env, n_train_episodes, seed, tb_writer=tb_writer)
+            tb_writer.close()
+
+            save_path = log_dir / "trainer.pkl"
+            save_trainer(trainer, save_path)
+
+            trained_trainers[run_name] = (trainer, env)
+
+    logger.info("Starting evaluation")
+
+    for run_name, (trainer, env) in trained_trainers.items():
+        logger.info(f"Evaluating {run_name}")
+        results = trainer.evaluate(n_episodes=n_eval_episodes, seed=seed + 1000)
+        for metric, (mean, std) in results.items():
+            logger.info(f"  {metric:20s}: {mean:10.2f} ± {std:6.2f}")
+
+    logger.info(f"Results saved to: {base_dir}")

tests/e2e/__init__.py

@@ -0,0 +1 @@
+"""E2E test suite for PHANTOM dynamic pricing pipeline."""

tests/e2e/fixtures.ts

@@ -0,0 +1,17 @@
+import { test as base } from '@playwright/test';
+
+type TestFixtures = {
+  backendUrl: string;
+  pricingUrl: string;
+};
+
+export const test = base.extend<TestFixtures>({
+  backendUrl: async ({}, use) => {
+    await use(process.env.BACKEND_URL || 'http://localhost:5000');
+  },
+  pricingUrl: async ({}, use) => {
+    await use(process.env.PRICING_PROVIDER_URL || 'http://localhost:5001');
+  },
+});
+
+export { expect } from '@playwright/test';

tests/e2e/helpers/api.ts

@@ -0,0 +1,69 @@
+interface PriceResponse {
+  price: number;
+  base_price: number;
+  markup: number;
+  model_version?: string;
+}
+
+export async function fetchPrice(
+  baseUrl: string,
+  productId: string,
+  mode: string = 'simple_surge',
+  sessionId?: string
+): Promise<PriceResponse> {
+  const params = new URLSearchParams();
+  if (sessionId) params.set('sessionId', sessionId);
+
+  const url = `${baseUrl}/api/pricing?mode=${mode}&productId=${productId}&${params}`;
+  const resp = await fetch(url);
+
+  if (!resp.ok) {
+    throw new Error(`Price fetch failed: ${resp.status}`);
+  }
+
+  return resp.json();
+}
+
+export async function waitForPriceChange(
+  baseUrl: string,
+  productId: string,
+  baselinePrice: number,
+  mode: string,
+  sessionId?: string,
+  maxRetries: number = 10,
+  pollInterval: number = 500
+): Promise<PriceResponse> {
+  for (let i = 0; i < maxRetries; i++) {
+    const priceResp = await fetchPrice(baseUrl, productId, mode, sessionId);
+    if (Math.abs(priceResp.price - baselinePrice) > 0.01) {
+      return priceResp;
+    }
+    await new Promise(r => setTimeout(r, pollInterval));
+  }
+
+  throw new Error(`Price did not change after ${maxRetries} retries`);
+}
+
+export async function ingestEvent(
+  baseUrl: string,
+  sessionId: string,
+  event: string,
+  productId?: string,
+  metadata?: Record<string, any>
+): Promise<void> {
+  const resp = await fetch(`${baseUrl}/api/ingest`, {
+    method: 'POST',
+    headers: { 'Content-Type': 'application/json' },
+    body: JSON.stringify({
+      sessionId,
+      event,
+      productId,
+      timestamp: new Date().toISOString(),
+      metadata,
+    }),
+  });
+
+  if (!resp.ok) {
+    throw new Error(`Event ingest failed: ${resp.status}`);
+  }
+}

tests/e2e/helpers/interactions.ts

@@ -0,0 +1,219 @@
+import { Page } from '@playwright/test';
+
+export async function getSessionId(page: Page): Promise<string | null> {
+  const cookies = await page.context().cookies();
+  const sessionCookie = cookies.find(c => c.name === 'phantom_session_id');
+  return sessionCookie?.value || null;
+}
+
+export async function verifySessionConsistency(page: Page, expectedSessionId: string): Promise<boolean> {
+  const currentSessionId = await getSessionId(page);
+  return currentSessionId === expectedSessionId;
+}
+
+export async function createFreshSession(page: Page, storeType: 'hotel' | 'airline' = 'hotel'): Promise<string> {
+  await page.context().clearCookies();
+  await page.goto('/');
+  await page.waitForLoadState('networkidle');
+  await page.waitForTimeout(500);
+
+  const sid = await getSessionId(page);
+  if (!sid) throw new Error('Session not created');
+  return sid;
+}
+
+interface SearchParams {
+  destination?: string;
+  checkIn?: string;
+  guests?: number;
+  rooms?: number;
+  origin?: string;
+  departure?: string;
+  adults?: number;
+}
+
+export async function performSearch(page: Page, params: SearchParams, storeType: 'hotel' | 'airline' = 'hotel' ): Promise<void> {
+  await page.waitForLoadState('networkidle');
+
+  if (storeType === 'hotel') {
+    const destInput = page.locator('input#destination');
+    await destInput.fill(params.destination || 'New York');
+
+    const checkInInput = page.locator('input#checkIn');
+    const checkInDate = params.checkIn || new Date(Date.now() + 7 * 86400000).toISOString().split('T')[0];
+    await checkInInput.fill(checkInDate);
+
+    const searchBtn = page.locator('button:has-text("Search Rooms")');
+    await searchBtn.click();
+  } else {
+    const originDropdown = page.locator('button:has-text("Select origin")').or(
+      page.locator('[id="origin"]').locator('button').first()
+    );
+    await originDropdown.click();
+    await page.waitForTimeout(200);
+    const originOption = page.locator(`button:has-text("${params.origin || 'JFK'}")`).first();
+    await originOption.click();
+    await page.waitForTimeout(200);
+
+    const destDropdown = page.locator('button:has-text("Select destination")').or(
+      page.locator('[id="destination"]').locator('button').first()
+    );
+    await destDropdown.click();
+    await page.waitForTimeout(200);
+    const destOption = page.locator(`button:has-text("${params.destination || 'LAX'}")`).first();
+    await destOption.click();
+    await page.waitForTimeout(200);
+
+    const departInput = page.locator('input#departDate');
+    const departDate = params.departure || new Date(Date.now() + 7 * 86400000).toISOString().split('T')[0];
+    await departInput.fill(departDate);
+
+    const searchBtn = page.locator('button:has-text("Search Flights")');
+    await searchBtn.click();
+  }
+
+  await page.waitForLoadState('networkidle');
+}
+
+export async function selectRandomProduct(page: Page, storeType: 'hotel' | 'airline' = 'hotel'): Promise<string> {
+  await page.waitForLoadState('networkidle');
+
+  const cardClass = storeType === 'hotel' ? '.hotel-card' : '.flight-card';
+  const productCards = page.locator(cardClass);
+
+  const count = await productCards.count();
+  if (count === 0) throw new Error('No products found on listing page');
+
+  const randomIdx = Math.floor(Math.random() * count);
+  return randomIdx.toString();
+}
+
+export async function openProductFromListing(page: Page, productId?: string): Promise<string> {
+  await page.waitForLoadState('networkidle');
+
+  const hotelCards = page.locator('.hotel-card');
+  const flightCards = page.locator('.flight-card');
+
+  const hotelCount = await hotelCards.count();
+  const flightCount = await flightCards.count();
+
+  let productCards;
+  if (hotelCount > 0) {
+    productCards = hotelCards;
+  } else if (flightCount > 0) {
+    productCards = flightCards;
+  } else {
+    throw new Error('No products found on listing page');
+  }
+
+  const count = await productCards.count();
+  const randomIdx = productId ? 0 : Math.floor(Math.random() * count);
+  await productCards.nth(randomIdx).click();
+
+  await page.waitForLoadState('networkidle');
+
+  const url = page.url();
+  const match = url.match(/\/products\/([^/?]+)/);
+  if (!match) throw new Error('Cannot parse product ID from URL after navigation');
+
+  return match[1];
+}
+
+export async function getPriceFromDOM(page: Page): Promise<number> {
+  await page.waitForLoadState('networkidle');
+
+  await page.waitForSelector('.price-amount', { timeout: 15000 }).catch(() => null);
+
+  const priceSelectors = [
+    '.price-amount',
+    '.price-display',
+    '[data-testid="price"]',
+    '[data-price]',
+  ];
+
+  for (const selector of priceSelectors) {
+    const priceEl = page.locator(selector).first();
+    if (await priceEl.count() > 0) {
+      const text = await priceEl.textContent();
+      if (!text) continue;
+
+      const match = text.match(/[\$]?\s*([\d,]+(?:\.\d{2})?)/);
+      if (match) {
+        const priceStr = match[1].replace(/,/g, '');
+        return parseFloat(priceStr);
+      }
+    }
+  }
+
+  const dataPrice = await page.locator('[data-price]').first().getAttribute('data-price').catch(() => null);
+  if (dataPrice) return parseFloat(dataPrice);
+
+  throw new Error('Cannot extract price from DOM');
+}
+
+export async function navigateToProduct(page: Page,productId: string,storeType: 'hotel' | 'airline' = 'hotel'): Promise<void> {
+  await page.goto(`/products/${productId}`);
+  await page.waitForLoadState('networkidle');
+}
+
+export async function viewProductViaFlow(page: Page, storeType: 'hotel' | 'airline' = 'hotel', searchParams?: SearchParams): Promise<string> {
+  const params = new URLSearchParams();
+  params.set('dateIndex', '7');
+
+  if (storeType === 'hotel') {
+    params.set('destination', searchParams?.destination || 'New York');
+    params.set('adults', '2');
+    params.set('rooms', '1');
+  } else {
+    params.set('origin', searchParams?.origin || 'JFK');
+    params.set('destination', searchParams?.destination || 'LAX');
+    params.set('adults', '1');
+    params.set('children', '0');
+    params.set('infants', '0');
+  }
+
+  await page.goto(`/products?${params.toString()}`);
+  await page.waitForLoadState('networkidle');
+
+  const productId = await openProductFromListing(page);
+  await page.waitForTimeout(500);
+  return productId;
+}
+
+export async function rapidViewProductViaFlow(page: Page, count: number, delayMs: number = 100, storeType: 'hotel' | 'airline' = 'hotel'): Promise<string[]> {
+  const productIds: string[] = [];
+
+  for (let i = 0; i < count; i++) {
+    const productId = await viewProductViaFlow(page, storeType);
+    productIds.push(productId);
+
+    await page.waitForTimeout(delayMs);
+  }
+
+  return productIds;
+}
+
+export async function humanLikeViewProduct(page: Page, storeType: 'hotel' | 'airline' = 'hotel'
+): Promise<string> {
+  const productId = await viewProductViaFlow(page, storeType);
+
+  await page.hover('h1');
+  await page.waitForTimeout(800 + Math.random() * 400);
+
+  await page.mouse.wheel(0, 200);
+  await page.waitForTimeout(500 + Math.random() * 300);
+
+  const paragraphs = await page.locator('p').all();
+  if (paragraphs.length > 0) {
+    await paragraphs[0].hover();
+    await page.waitForTimeout(600 + Math.random() * 400);
+  }
+
+  return productId;
+}
+
+export async function addToCart(page: Page): Promise<void> {
+  const addBtn = page.locator('button:has-text("Add to Cart")');
+  await addBtn.click();
+  await page.waitForTimeout(500);
+}

tests/e2e/helpers/kafka.ts

@@ -0,0 +1,39 @@
+interface InteractionEvent {
+  sessionId: string;
+  event: string;
+  productId?: string;
+  timestamp: string;
+  metadata?: Record<string, any>;
+}
+
+const dumpKafkaTopic = async (backendUrl: string, topic: string) => {
+  const resp = await fetch(`${backendUrl}/api/kafka/dump?topic=${topic}`);
+  if (!resp.ok) throw new Error(`Kafka dump failed: ${resp.status}`);
+  const { data = [] } = await resp.json();
+  return data as any[];
+};
+
+export const waitForInteractionEvent = async (
+  backendUrl: string,
+  sessionId: string,
+  eventType: string,
+  maxRetries = 10,
+  pollInterval = 500
+): Promise<InteractionEvent | null> => {
+  for (let i = 0; i < maxRetries; i++) {
+    const msgs = await dumpKafkaTopic(backendUrl, "user-interactions");
+    const hit = msgs.find(m => m.sessionId === sessionId && m.event === eventType);
+    if (hit) return hit as InteractionEvent;
+    await new Promise<void>(r => setTimeout(r, pollInterval));
+  }
+  return null;
+};
+
+export const countProductViews = async (backendUrl: string, productId: string) =>
+  (await dumpKafkaTopic(backendUrl, "user-interactions")).reduce(
+    (n, m) => n + (m.productId === productId && m.event === "view_item_page" ? 1 : 0),
+    0
+  );
+
+export const getSessionEvents = async (backendUrl: string, sessionId: string) =>
+  (await dumpKafkaTopic(backendUrl, "user-interactions")).filter(m => m.sessionId === sessionId);

tests/e2e/package.json

@@ -0,0 +1,19 @@
+{
+  "name": "e2e",
+  "version": "1.0.0",
+  "main": "index.js",
+  "scripts": {
+    "test": "playwright test",
+    "test:ui": "playwright test --ui",
+    "test:debug": "playwright test --debug"
+  },
+  "keywords": [],
+  "author": "",
+  "license": "ISC",
+  "description": "",
+  "devDependencies": {
+    "@playwright/test": "^1.57.0",
+    "@types/node": "^25.0.6",
+    "typescript": "^5.9.3"
+  }
+}

tests/e2e/playwright.config.ts

@@ -0,0 +1,25 @@
+import { defineConfig, devices } from '@playwright/test';
+
+export default defineConfig({
+  testDir: './scenarios',
+  fullyParallel: true,
+  forbidOnly: !!process.env.CI,
+  retries: 0,
+  workers: 1,
+  reporter: 'list',
+  use: {
+    baseURL: process.env.WEB_URL || 'http://localhost:3000',
+    trace: 'retain-on-failure',
+    screenshot: 'only-on-failure',
+  },
+  timeout: 180000,
+  expect: {
+    timeout: 10000,
+  },
+  projects: [
+    {
+      name: 'chromium',
+      use: { ...devices['Desktop Chrome'] },
+    },
+  ],
+});

tests/e2e/scenarios/session-aware.spec.ts

@@ -0,0 +1,163 @@
+import { test, expect } from '../fixtures';
+import {
+  createFreshSession,
+  viewProductViaFlow,
+  rapidViewProductViaFlow,
+  humanLikeViewProduct,
+  getPriceFromDOM,
+  verifySessionConsistency,
+  addToCart,
+} from '../helpers/interactions';
+import { getSessionEvents } from '../helpers/kafka';
+import { runSessionPricing } from '../helpers/airflow';
+
+test.describe('SessionAwarePricer E2E', () => {
+  const STORE_TYPE = 'hotel';
+
+  test('baseline: human-like behavior maintains base price', async ({ page, backendUrl }) => {
+    const sessionId = await createFreshSession(page, STORE_TYPE);
+
+    const productId1 = await humanLikeViewProduct(page, STORE_TYPE);
+    const baselinePrice = await getPriceFromDOM(page);
+    expect(await verifySessionConsistency(page, sessionId)).toBeTruthy();
+
+    await page.waitForTimeout(1500);
+
+    const productId2 = await humanLikeViewProduct(page, STORE_TYPE);
+
+    await runSessionPricing(STORE_TYPE);
+
+    const secondPrice = await getPriceFromDOM(page);
+    expect(await verifySessionConsistency(page, sessionId)).toBeTruthy();
+
+    expect(Math.abs(secondPrice - baselinePrice) / baselinePrice).toBeLessThan(0.1);
+  });
+
+  test('agent detection: rapid robot-like behavior increases price', async ({ page, backendUrl }) => {
+    const sessionId = await createFreshSession(page, STORE_TYPE);
+
+    const productId = await viewProductViaFlow(page, STORE_TYPE);
+    const baselinePrice = await getPriceFromDOM(page);
+
+    await page.waitForTimeout(500);
+
+    await rapidViewProductViaFlow(page, 8, 100, STORE_TYPE);
+    expect(await verifySessionConsistency(page, sessionId)).toBeTruthy();
+
+    await page.waitForTimeout(1000);
+
+    const events = await getSessionEvents(backendUrl, sessionId);
+    expect(events.length).toBeGreaterThanOrEqual(8);
+
+    await runSessionPricing(STORE_TYPE);
+
+    await page.goto(`/products/${productId}`);
+    await page.waitForLoadState('networkidle');
+    const agentPrice = await getPriceFromDOM(page);
+
+    expect(agentPrice).toBeGreaterThan(baselinePrice);
+    expect((agentPrice - baselinePrice) / baselinePrice).toBeGreaterThan(0.01);
+  });
+
+  test('velocity threshold: high event rate triggers detection', async ({ page, backendUrl }) => {
+    const sessionId = await createFreshSession(page, STORE_TYPE);
+
+    const productId = await viewProductViaFlow(page, STORE_TYPE);
+    const baselinePrice = await getPriceFromDOM(page);
+
+    await rapidViewProductViaFlow(page, 10, 80, STORE_TYPE);
+
+    const events = await getSessionEvents(backendUrl, sessionId);
+    expect(events.length).toBeGreaterThanOrEqual(10);
+
+    await runSessionPricing(STORE_TYPE);
+
+    await page.goto(`/products/${productId}`);
+    await page.waitForLoadState('networkidle');
+    const agentPrice = await getPriceFromDOM(page);
+
+    expect(agentPrice).toBeGreaterThan(baselinePrice);
+    expect(await verifySessionConsistency(page, sessionId)).toBeTruthy();
+  });
+
+  test('cart ratio: high cart/view ratio signals intent', async ({ page, backendUrl }) => {
+    const sessionId = await createFreshSession(page, STORE_TYPE);
+
+    const productId = await viewProductViaFlow(page, STORE_TYPE);
+    const baselinePrice = await getPriceFromDOM(page);
+
+    await page.waitForTimeout(500);
+    await addToCart(page);
+
+    await page.waitForTimeout(2000);
+
+    await page.goto(`/products/${productId}`);
+    await page.waitForLoadState('networkidle');
+    const cartPrice = await getPriceFromDOM(page);
+
+    expect(cartPrice).toBeGreaterThanOrEqual(baselinePrice);
+    expect(await verifySessionConsistency(page, sessionId)).toBeTruthy();
+  });
+
+  test('mixed behavior: occasional fast actions tolerated', async ({ page, backendUrl }) => {
+    const sessionId = await createFreshSession(page, STORE_TYPE);
+
+    const productId1 = await humanLikeViewProduct(page, STORE_TYPE);
+    const baselinePrice = await getPriceFromDOM(page);
+
+    await page.waitForTimeout(1200);
+
+    await rapidViewProductViaFlow(page, 2, 150, STORE_TYPE);
+
+    await page.waitForTimeout(1000);
+    await humanLikeViewProduct(page, STORE_TYPE);
+
+    await runSessionPricing(STORE_TYPE);
+
+    const finalPrice = await getPriceFromDOM(page);
+
+    expect(Math.abs(finalPrice - baselinePrice) / baselinePrice).toBeLessThan(0.3);
+    expect(await verifySessionConsistency(page, sessionId)).toBeTruthy();
+  });
+
+  test('session isolation: agent behavior in one session does not affect others', async ({
+    page,
+    context,
+    backendUrl,
+  }) => {
+    const sessionIdA = await createFreshSession(page, STORE_TYPE);
+    const productId = await viewProductViaFlow(page, STORE_TYPE);
+    const basePrice = await getPriceFromDOM(page);
+
+    await rapidViewProductViaFlow(page, 10, 100, STORE_TYPE);
+    await page.waitForTimeout(2000);
+
+    await page.goto(`/products/${productId}`);
+    await page.waitForLoadState('networkidle');
+    const agentPrice = await getPriceFromDOM(page);
+    expect(agentPrice).toBeGreaterThan(basePrice * 0.99);
+
+    const page2 = await context.newPage();
+    const sessionIdB = await createFreshSession(page2, STORE_TYPE);
+
+    await page2.goto(`/products/${productId}`);
+    await page2.waitForLoadState('networkidle');
+    const cleanPrice = await getPriceFromDOM(page2);
+
+    expect(Math.abs(cleanPrice - basePrice) / basePrice).toBeLessThan(0.1);
+    expect(sessionIdA).not.toBe(sessionIdB);
+  });
+
+  test('session persistence: session ID maintained across views', async ({ page }) => {
+    const sessionId = await createFreshSession(page, STORE_TYPE);
+
+    await viewProductViaFlow(page, STORE_TYPE);
+    expect(await verifySessionConsistency(page, sessionId)).toBeTruthy();
+
+    await viewProductViaFlow(page, STORE_TYPE);
+    expect(await verifySessionConsistency(page, sessionId)).toBeTruthy();
+
+    await viewProductViaFlow(page, STORE_TYPE);
+    expect(await verifySessionConsistency(page, sessionId)).toBeTruthy();
+  });
+});

tests/e2e/scenarios/surge-pricing.spec.ts

@@ -0,0 +1,118 @@
+import { test, expect } from '../fixtures';
+import {
+  createFreshSession,
+  viewProductViaFlow,
+  rapidViewProductViaFlow,
+  getPriceFromDOM,
+  verifySessionConsistency,
+} from '../helpers/interactions';
+import { waitForInteractionEvent, countProductViews } from '../helpers/kafka';
+import { runSurgePricing } from '../helpers/airflow';
+
+test.describe('SimpleSurgePricer E2E', () => {
+  const STORE_TYPE = 'hotel';
+
+  test('baseline: initial price equals base price', async ({ page, backendUrl }) => {
+    const sessionId = await createFreshSession(page, STORE_TYPE);
+
+    const productId = await viewProductViaFlow(page, STORE_TYPE);
+    const price = await getPriceFromDOM(page);
+
+    expect(price).toBeGreaterThan(0);
+    expect(await verifySessionConsistency(page, sessionId)).toBeTruthy();
+  });
+
+  test('surge: rapid views trigger price increase', async ({ page, backendUrl }) => {
+    const sessionId = await createFreshSession(page, STORE_TYPE);
+
+    const productId = await viewProductViaFlow(page, STORE_TYPE);
+    const baselinePrice = await getPriceFromDOM(page);
+
+    await rapidViewProductViaFlow(page, 5, 200, STORE_TYPE);
+
+    await page.waitForTimeout(1000);
+
+    const evt = await waitForInteractionEvent(backendUrl, sessionId, 'view_item_page');
+    expect(evt).not.toBeNull();
+
+    const viewCount = await countProductViews(backendUrl, productId);
+    expect(viewCount).toBeGreaterThanOrEqual(5);
+
+    await runSurgePricing(STORE_TYPE, 3, 1);
+
+    await page.goto(`/products/${productId}`);
+    await page.waitForLoadState('networkidle');
+    const surgedPrice = await getPriceFromDOM(page);
+
+    expect(surgedPrice).toBeGreaterThan(baselinePrice);
+    expect((surgedPrice - baselinePrice) / baselinePrice).toBeGreaterThan(0.01);
+    expect(await verifySessionConsistency(page, sessionId)).toBeTruthy();
+  });
+
+  test('threshold: price unchanged below threshold', async ({ page, backendUrl }) => {
+    const sessionId = await createFreshSession(page, STORE_TYPE);
+
+    const productId = await viewProductViaFlow(page, STORE_TYPE);
+    const baselinePrice = await getPriceFromDOM(page);
+
+    await rapidViewProductViaFlow(page, 2, 300, STORE_TYPE);
+
+    await page.waitForTimeout(1500);
+
+    await page.goto(`/products/${productId}`);
+    await page.waitForLoadState('networkidle');
+    const currentPrice = await getPriceFromDOM(page);
+
+    expect(Math.abs(currentPrice - baselinePrice) / baselinePrice).toBeLessThan(0.05);
+    expect(await verifySessionConsistency(page, sessionId)).toBeTruthy();
+  });
+
+  test('window: surge decays after window expires', async ({ page, backendUrl }) => {
+    const sessionId = await createFreshSession(page, STORE_TYPE);
+
+    const productId = await viewProductViaFlow(page, STORE_TYPE);
+    const baselinePrice = await getPriceFromDOM(page);
+
+    await rapidViewProductViaFlow(page, 5, 150, STORE_TYPE);
+
+    await page.waitForTimeout(1000);
+
+    await runSurgePricing(STORE_TYPE, 3, 1);
+
+    await page.goto(`/products/${productId}`);
+    await page.waitForLoadState('networkidle');
+    const surgedPrice = await getPriceFromDOM(page);
+    expect(surgedPrice).toBeGreaterThan(baselinePrice);
+
+    await page.waitForTimeout(12000);
+
+    await runSurgePricing(STORE_TYPE, 3, 1);
+
+    await page.goto(`/products/${productId}`);
+    await page.waitForLoadState('networkidle');
+    const decayedPrice = await getPriceFromDOM(page);
+    expect(decayedPrice).toBeLessThan(surgedPrice);
+    expect(await verifySessionConsistency(page, sessionId)).toBeTruthy();
+  });
+
+  test('isolation: different products have independent surge', async ({ page, backendUrl }) => {
+    const sessionId = await createFreshSession(page, STORE_TYPE);
+
+    const productIdA = await viewProductViaFlow(page, STORE_TYPE);
+    const basePriceA = await getPriceFromDOM(page);
+
+    await rapidViewProductViaFlow(page, 5, 200, STORE_TYPE);
+    await page.waitForTimeout(2000);
+
+    await page.goto(`/products/${productIdA}`);
+    await page.waitForLoadState('networkidle');
+    const surgedPriceA = await getPriceFromDOM(page);
+
+    const productIdB = await viewProductViaFlow(page, STORE_TYPE);
+    const priceB = await getPriceFromDOM(page);
+
+    expect(surgedPriceA).toBeGreaterThan(basePriceA * 0.99);
+    expect(productIdA).not.toBe(productIdB);
+    expect(await verifySessionConsistency(page, sessionId)).toBeTruthy();
+  });
+});

tests/e2e/tsconfig.json

@@ -0,0 +1,15 @@
+{
+  "compilerOptions": {
+    "target": "ES2022",
+    "module": "commonjs",
+    "lib": ["ES2022"],
+    "strict": true,
+    "esModuleInterop": true,
+    "skipLibCheck": true,
+    "forceConsistentCasingInFileNames": true,
+    "resolveJsonModule": true,
+    "types": ["node", "@playwright/test"]
+  },
+  "include": ["**/*.ts"],
+  "exclude": ["node_modules"]
+}

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

@@ -30,6 +30,8 @@ export async function GET(req: NextRequest) {
     const providerUrl = process.env.PRICING_PROVIDER_URL || 'http://localhost:5001';
     try {
         const queryParams = new URLSearchParams();
+        // THIS is our entry point into the dynamic pricing where we reference the context of the sesion and experiment and ask for a price to assign to the trajectory which is expressed
+        // The whole pipeline gets triggered from here.
         if (sessionId) queryParams.append('sessionId', sessionId);
         if (experimentId) queryParams.append('experimentId', experimentId);
 
@@ -55,11 +57,11 @@ export async function GET(req: NextRequest) {
         price = Math.round(randomBase * 100) / 100;
     }
 
-    // log price to kafka for elasticity computation
+    // log price to kafka asynchronously (non-blocking)
     if (sessionId) {
         const backendUrl = process.env.BACKEND_URL || 'http://localhost:5000';
-        try {
+        // fire and forget - don't await to avoid blocking response
-            await fetch(`${backendUrl}/api/kafka/price-log`, {
+        fetch(`${backendUrl}/api/kafka/price-log`, {
             method: 'POST',
             headers: { 'Content-Type': 'application/json' },
             body: JSON.stringify({
@@ -70,10 +72,11 @@ export async function GET(req: NextRequest) {
                 storeMode,
                 ts: timestamp,
             }),
-            });
+        }).catch(err => {
-        } catch (err) {
+            if (process.env.NODE_ENV === 'development') {
                 console.error('[price-log-error]', err);
             }
+        });
     }
 
     if (process.env.NODE_ENV === 'development') {

web/src/app/cart/page.tsx

@@ -32,7 +32,8 @@ export default function CartPage() {
                     {itemCount > 0 && (
                         <button
                             onClick={clearCart}
-                            className="text-sm text-red-600 hover:underline"
+                            className="text-sm hover:underline"
+                            style={{ color: 'var(--accent-warning)' }}
                         >
                             Clear cart
                         </button>
@@ -42,7 +43,7 @@ export default function CartPage() {
                 {itemCount === 0 ? (
                     <div className="text-center py-12">
                         <p className="text-gray-500 mb-4">Your cart is empty</p>
-                        <a href="/" className="text-blue-600 hover:underline">Browse our selection</a>
+                        <a href="/" className="hover:underline" style={{ color: 'var(--text-accent)' }}>Browse our selection</a>
                     </div>
                 ) : (
                     <>
@@ -54,15 +55,11 @@ export default function CartPage() {
                                 >
                                     <div className="flex-1">
                                         <div className="flex items-center gap-2 mb-1">
-                                            <span className="px-2 py-0.5 text-xs font-medium rounded bg-blue-100 text-blue-800">
-                                                {item.type}
-                                            </span>
                                             <h3 className="font-semibold">{item.name}</h3>
                                         </div>
 
                                         {item.type === 'hotel' && (
                                             <div className="text-sm text-gray-600">
-                                                <p>{String(item.metadata.roomType)}</p>
                                                 <p>{String(item.metadata.checkIn)} - {String(item.metadata.checkOut)}</p>
                                                 <p>{String(item.metadata.nights)} night{Number(item.metadata.nights) > 1 ? 's' : ''}</p>
                                             </div>
@@ -81,7 +78,8 @@ export default function CartPage() {
                                         <p className="text-xl font-bold mb-2">${item.price}</p>
                                         <button
                                             onClick={() => handleRemove(item.id, item.type)}
-                                            className="text-sm text-red-600 hover:underline"
+                                            className="text-sm hover:underline"
+                                            style={{ color: 'var(--accent-warning)' }}
                                         >
                                             Remove
                                         </button>
@@ -100,7 +98,7 @@ export default function CartPage() {
                                     dispatchInteraction('checkout_start', undefined, { total, itemCount });
                                     window.location.href = '/checkout';
                                 }}
-                                className="w-full py-3 bg-blue-600 hover:bg-blue-700 text-white rounded-lg font-medium transition-colors"
+                                className="btn-primary w-full"
                             >
                                 Proceed to Checkout
                             </button>

web/src/app/globals.css

@@ -8,6 +8,9 @@
   --bg-secondary: #f5f5f5;
   --text-primary: #333333;
   --text-secondary: #666666;
+  --accent-primary: #007aff;
+  --accent-primary-hover: #0051d5;
+  --accent-primary-light: #e6f2ff;
   --spacing-sm: 8px;
   --spacing-md: 16px;
   --spacing-lg: 32px;

web/src/app/layout.tsx

@@ -15,8 +15,8 @@ const geistMono = Geist_Mono({
 });
 
 export const metadata: Metadata = {
-  title: "Create Next App",
+  title: "Travel Booking Platform",
-  description: "Generated by create next app",
+  description: "Book flights and hotels with dynamic pricing",
 };
 
 export default function RootLayout({

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

@@ -2,6 +2,7 @@
 
 import type { EventName } from '@/lib/events';
 import type { Hotel } from '@/lib/hotel-utils';
+import { getHotelImageUrl } from '@/lib/hotel-utils';
 import { useHoverTracking } from '@/hooks/useHoverTracking';
 import PriceDisplay from '@/components/ui/PriceDisplay';
 
@@ -47,8 +48,6 @@ export default function HotelCard({ hotel }: { hotel: Hotel }) {
         window.location.href = `/hotel/products/${hotel.id}`;
     };
 
-    const imageUrl = `https://images.unsplash.com/photo-1551882547-ff40c63fe5fa?w=400&h=300&fit=crop`;
-
     return (
         <div
             className="hotel-card cursor-pointer"
@@ -56,7 +55,7 @@ export default function HotelCard({ hotel }: { hotel: Hotel }) {
         >
             <div className="hotel-image relative overflow-hidden">
                 <img
-                    src={imageUrl}
+                    src={getHotelImageUrl(hotel.id, { w: 400, h: 300 })}
                     alt={hotel.name}
                     className="w-full h-full object-cover"
                     onError={(e) => {

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

@@ -2,6 +2,7 @@
 
 import { useState, useEffect } from 'react';
 import type { Hotel } from '@/lib/hotel-utils';
+import { getHotelImageUrl } from '@/lib/hotel-utils';
 import PriceDisplay from '@/components/ui/PriceDisplay';
 
 interface HotelDetailsProps {
@@ -43,13 +44,11 @@ const PriceTotalDisplay = ({ productId, nights }: { productId: string; nights: n
 };
 
 export default function HotelDetails({ product, onAddToCart, addedToCart }: HotelDetailsProps) {
-  const imageUrl = `https://images.unsplash.com/photo-1566073771259-6a8506099945?w=800&h=600&fit=crop`;
-
   return (
     <div className="w-full flex flex-col lg:flex-row gap-12 py-8">
       <div className="w-full lg:w-1/2 rounded-lg aspect-[4/3] overflow-hidden shrink-0">
         <img
-          src={imageUrl}
+          src={getHotelImageUrl(product.id, { w: 800, h: 600 })}
           alt={product.name}
           className="w-full h-full object-cover"
           onError={(e) => {

web/src/components/ui/Navigation.tsx

@@ -20,7 +20,7 @@ const NavLink = ({ href, children }: { href: string; children: React.ReactNode }
       href={href}
       className={`px-4 py-2 rounded-md transition-colors ${
         isActive
-          ? 'bg-[var(--accent-primary)] font-semibold'
+          ? 'bg-[var(--accent-primary)] text-white font-semibold'
           : 'hover:bg-[var(--accent-primary-light)] text-[var(--text-primary)]'
       }`}
     >

web/src/lib/airline-utils.ts

@@ -31,7 +31,7 @@ export interface Flight {
   availability: number;
 }
 
-const EPOCH = new Date(0);
+import { dateToDaysFromToday, dateToIndex, todayIndex } from './date-utils';
 
 export const transformProduct = (p: AirlineProduct): Flight => {
   const { id, flight_type, date_index, metadata, availability } = p;
@@ -52,24 +52,4 @@ export const transformProduct = (p: AirlineProduct): Flight => {
   };
 };
 
-// convert date string to days from today
+export { dateToDaysFromToday, dateToIndex, todayIndex };
-export const dateToDaysFromToday = (dateStr: string): number => {
-  const target = new Date(dateStr);
-  target.setHours(0, 0, 0, 0);
-  const today = new Date();
-  today.setHours(0, 0, 0, 0);
-  return Math.floor((target.getTime() - today.getTime()) / 86400000);
-};
-
-// convert date string to date_index (days since epoch)
-export const dateToIndex = (dateStr: string): number => {
-  const d = new Date(dateStr);
-  return Math.floor((d.getTime() - EPOCH.getTime()) / 86400000);
-};
-
-// get current date_index
-export const todayIndex = (): number => {
-  const now = new Date();
-  now.setHours(0, 0, 0, 0);
-  return Math.floor((now.getTime() - EPOCH.getTime()) / 86400000);
-};

web/src/lib/date-utils.ts

@@ -0,0 +1,23 @@
+const EPOCH = new Date(0);
+const MS_PER_DAY = 86400000;
+
+export const dateToDaysFromToday = (dateStr: string): number => {
+  const target = new Date(dateStr);
+  target.setHours(0, 0, 0, 0);
+  const today = new Date();
+  today.setHours(0, 0, 0, 0);
+  return Math.floor((target.getTime() - today.getTime()) / MS_PER_DAY);
+};
+
+export const dateToIndex = (dateStr: string): number => {
+  const d = new Date(dateStr);
+  return Math.floor((d.getTime() - EPOCH.getTime()) / MS_PER_DAY);
+};
+
+export const todayIndex = (): number => {
+  const now = new Date();
+  now.setHours(0, 0, 0, 0);
+  return Math.floor((now.getTime() - EPOCH.getTime()) / MS_PER_DAY);
+};
+
+export { EPOCH, MS_PER_DAY };

web/src/lib/hotel-utils.ts

@@ -25,7 +25,7 @@ export interface Hotel {
   nights: number;
 }
 
-const EPOCH = new Date(0);
+import { EPOCH, MS_PER_DAY, dateToDaysFromToday, dateToIndex, todayIndex } from './date-utils';
 
 export const transformProduct = (p: HotelProduct): Hotel => {
   const { id, room_type, date_index, metadata } = p;
@@ -37,14 +37,14 @@ export const transformProduct = (p: HotelProduct): Hotel => {
     // legacy: treat as offset from today
     const today = new Date();
     today.setHours(0, 0, 0, 0);
-    checkIn = new Date(today.getTime() + date_index * 86400000);
+    checkIn = new Date(today.getTime() + date_index * MS_PER_DAY);
   } else {
     // proper: days since epoch
-    checkIn = new Date(EPOCH.getTime() + date_index * 86400000);
+    checkIn = new Date(EPOCH.getTime() + date_index * MS_PER_DAY);
   }
 
   const nights = 1;
-  const checkOut = new Date(checkIn.getTime() + nights * 86400000);
+  const checkOut = new Date(checkIn.getTime() + nights * MS_PER_DAY);
 
   const formatOpts: Intl.DateTimeFormatOptions = {
     month: 'short',
@@ -65,24 +65,34 @@ export const transformProduct = (p: HotelProduct): Hotel => {
   };
 };
 
-// convert date string to days from today
+const hotelImagePool = [
-export const dateToDaysFromToday = (dateStr: string): number => {
+  'photo-1566073771259-6a8506099945',
-  const target = new Date(dateStr);
+  'photo-1551882547-ff40c63fe5fa',
-  target.setHours(0, 0, 0, 0);
+  'photo-1590490360182-c33d57733427',
-  const today = new Date();
+  'photo-1582719478250-c89cae4dc85b',
-  today.setHours(0, 0, 0, 0);
+  'photo-1596701062351-8c2c14d1fdd0',
-  return Math.floor((target.getTime() - today.getTime()) / 86400000);
+  'photo-1631049307264-da0ec9d70304',
+  'photo-1578683010236-d716f9a3f461',
+  'photo-1540518614846-7eded433c457',
+  'photo-1505693416388-ac5ce068fe85',
+  'photo-1522771739844-6a9f6d5f14af',
+  'photo-1562438668-bcf0ca6578f0',
+  'photo-1595576508898-0ad5c879a061',
+];
+
+const hashString = (s: string): number => {
+  let h = 0;
+  for (let i = 0; i < s.length; i++) {
+    h = ((h << 5) - h) + s.charCodeAt(i);
+    h = h & h;
+  }
+  return Math.abs(h);
 };
 
-// convert date string to date_index (days since epoch)
+export const getHotelImageUrl = (hotelId: string, size: { w: number; h: number } = { w: 400, h: 300 }): string => {
-export const dateToIndex = (dateStr: string): number => {
+  const idx = hashString(hotelId) % hotelImagePool.length;
-  const d = new Date(dateStr);
+  const photoId = hotelImagePool[idx];
-  return Math.floor((d.getTime() - EPOCH.getTime()) / 86400000);
+  return `https://images.unsplash.com/${photoId}?w=${size.w}&h=${size.h}&fit=crop`;
 };
 
-// get current date_index
+export { dateToDaysFromToday, dateToIndex, todayIndex };
-export const todayIndex = (): number => {
-  const now = new Date();
-  now.setHours(0, 0, 0, 0);
-  return Math.floor((now.getTime() - EPOCH.getTime()) / 86400000);
-};