chore: refactoring function definition to avoid reinit

Co-authored-by: gemini-code-assist[bot] <176961590+gemini-code-assist[bot]@users.noreply.github.com>

.gitignore

@@ -11,6 +11,3 @@ 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,74 +11,46 @@ PYTEST    := $(VENV)/bin/pytest
 
 .DEFAULT_GOAL := help
 
-.PHONY: help
+all: pdf
-help:
+
-	@echo "pdf.build pdf.watch pdf.clean | test.backend test.e2e test.all | web.dev | install | stats.lines"
+run.webapp:
+	@cd web && npm install && npm run dev
 
 $(BUILDDIR):
 	mkdir -p paper/$(BUILDDIR)
 
-.PHONY: pdf.build
+pdf: $(BUILDDIR)
-pdf.build: $(BUILDDIR)
+	@echo "Concatenating source code..."
 	@bash paper/concat_code.sh
 	@cd $(SRCDIR) && \
 	$(LATEXMK) -pdf -jobname=$(JOBNAME) \
 		-interaction=nonstopmode -file-line-error \
 		-outdir=../$(BUILDDIR) $(TEX)
 
-.PHONY: pdf.watch
+watch: $(BUILDDIR)
-pdf.watch: $(BUILDDIR)
 	@cd $(SRCDIR) && \
 	$(LATEXMK) -pvc -pdf -jobname=$(JOBNAME) \
 		-interaction=nonstopmode -file-line-error \
 		-r ../.latexmkrc \
 		-outdir=../$(BUILDDIR) $(TEX)
 
-.PHONY: pdf.clean
+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
 
-.PHONY: stats.lines
+test: $(VENV)
-stats.lines:
+	$(PYTEST) -v
+
+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: pdf clean watch run.webapp test count-lines all
+.PHONY: all pdf clean watch run.webapp install test
-pdf: pdf.build
-clean: pdf.clean
-watch: pdf.watch
-run.webapp: web.dev
-test: test.backend
-count-lines: stats.lines
-all: pdf.build

backend/provider/app.py

@@ -47,52 +47,53 @@ 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)
 
-    # PRIORITY 1: session-aware price (computed by Airflow worker)
+    # fetch pre-computed prices from registry
-    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')
-    if prices_df is not None:
+    elasticity_df = registry.get_elasticity('latest')
-        product_price_row = prices_df[prices_df['productId'] == productId]
-        if not product_price_row.empty:
-            optimal_price = float(product_price_row['optimal_price'].iloc[0])
-            return PriceResponse(
-                productId=productId,
-                price=optimal_price,
-                base_price=base_price,
-                markup=optimal_price/base_price,
-                elasticity=None,
-                model_version='surge'
-            )
 
-    # PRIORITY 3: fallback to base price
+    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,
+            elasticity=None
-        model_version='base'
+        )
+
+    # lookup pre-computed price for this product
+    product_price_row = prices_df[prices_df['productId'] == productId]
+    if product_price_row.empty:
+        # product not in pre-computed prices, fallback to base
+        return PriceResponse(
+            productId=productId,
+            price=base_price,
+            base_price=base_price,
+            markup=1.0,
+            elasticity=None
+        )
+
+    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
     )
 
 @app.get("/models")

backend/server/app.py

@@ -198,16 +198,12 @@ def dump_logs(
             auto_offset_reset='earliest',
             enable_auto_commit=False,
             value_deserializer=lambda x: json.loads(x.decode('utf-8')),
-            consumer_timeout_ms=30000,
+            consumer_timeout_ms=5000
-            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,24 +1,4 @@
 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:
@@ -144,7 +124,6 @@ services:
       - AIRFLOW__CORE__ENABLE_XCOM_PICKLING=true
       - 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
@@ -181,7 +160,6 @@ services:
       - AIRFLOW__CORE__LOAD_EXAMPLES=false
       - AIRFLOW__CORE__ENABLE_XCOM_PICKLING=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

experiments/airflow/dags/ml_training_pipeline.py

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

experiments/airflow/dags/surge_pricing_pipeline.py

@@ -120,31 +120,15 @@ 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=high_thresh,
+        high_threshold=dag_conf.get('high_threshold', 10),
-        low_threshold=low_thresh,
+        low_threshold=dag_conf.get('low_threshold', 2),
-        surge_multiplier=surge_mult,
+        surge_multiplier=dag_conf.get('surge_multiplier', 1.2),
-        discount_multiplier=discount_mult
+        discount_multiplier=dag_conf.get('discount_multiplier', 0.9)
     )
     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

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

experiments/ml/arch.py

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

experiments/ml/evals.py

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

experiments/ml/requirements.txt

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

experiments/ml/train.py

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

experiments/procesing/pipelines.py

@@ -170,5 +170,3 @@ if __name__ == '__main__':
     print(f"Feature matrix: {features.shape}")
     print(features.head())
     print(features.info())
-
-    features.to_parquet("features.parquet")

experiments/procesing/pricers/simple.py

@@ -3,46 +3,6 @@ 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"""
 
@@ -107,25 +67,21 @@ class SimpleSurgePricer(PricingFunction):
         self.surge_multiplier = surge_multiplier
         self.discount_multiplier = discount_multiplier
 
-    def fit(self, market_data: pd.DataFrame):
+    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
-        return self
+        self.demand_history = market_data['demand'].to_numpy() if 'demand' in market_data.columns else np.zeros_like(self.base_prices)
 
-    def predict(self, state_space) -> np.ndarray:
+    def predict(self) -> np.ndarray:
         """
         Adjust prices based on current demand using surge rules.
-        state_space.demand: demand proxy per product (from session features)
+        state_space.demand: demand counts per product
-        state_space.prices: base prices
+        state_space.prices: current prices (fallback if base_prices not set)
         """
-        demand = np.asarray(state_space.demand) if state_space and hasattr(state_space, 'demand') else np.array([0])
+        current_prices = self.base_prices if self.base_prices is not None else np.ones_like(demand_vector) * 99.99
-        base = np.asarray(state_space.prices) if state_space and hasattr(state_space, 'prices') else self.base_prices
+        demand = self.demand_history if self.demand_history is not None else np.zeros_like(current_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
 

experiments/procesing/steps/session.py

@@ -135,7 +135,6 @@ class ExtractSessionFeaturesStep(BaseContextStep):
     Vectorized session feature extraction - replaces O(n^2) per-row loop.
     Input: interactions_df
     Output: session-level feature matrix
-    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, X: pd.DataFrame) -> pd.DataFrame:

lib/model_registry.py

@@ -178,49 +178,3 @@ 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

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

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

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

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

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

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

tests/e2e/fixtures.ts

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

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

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

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

@@ -1,19 +0,0 @@
-{
-  "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

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

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

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

@@ -1,15 +0,0 @@
-{
-  "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,8 +30,6 @@ 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);
 
@@ -57,11 +55,11 @@ export async function GET(req: NextRequest) {
         price = Math.round(randomBase * 100) / 100;
     }
 
-    // log price to kafka asynchronously (non-blocking)
+    // log price to kafka for elasticity computation
     if (sessionId) {
         const backendUrl = process.env.BACKEND_URL || 'http://localhost:5000';
-        // fire and forget - don't await to avoid blocking response
+        try {
-        fetch(`${backendUrl}/api/kafka/price-log`, {
+            await fetch(`${backendUrl}/api/kafka/price-log`, {
                 method: 'POST',
                 headers: { 'Content-Type': 'application/json' },
                 body: JSON.stringify({
@@ -72,11 +70,10 @@ export async function GET(req: NextRequest) {
                     storeMode,
                     ts: timestamp,
                 }),
-        }).catch(err => {
+            });
-            if (process.env.NODE_ENV === 'development') {
+        } catch (err) {
             console.error('[price-log-error]', err);
         }
-        });
     }
 
     if (process.env.NODE_ENV === 'development') {