chore: modules and reqs

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

README.md

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

docker-compose.yml

@@ -1,4 +1,15 @@
 services:
+
+  tensorboard:
+    image: tensorflow/tensorflow:latest
+    container_name: "PHANTOM-tensorboard"
+    ports:
+      - "6006:6006"
+    volumes:
+      - ./experiments/ml/runs:/logs
+    command: tensorboard --logdir=/logs --host=0.0.0.0 --port=6006
+    restart: unless-stopped
+
   backend:
     container_name: "PHANTOM-backend"
     build:

experiments/airflow/dags/ml_training_pipeline.py

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

experiments/ml/__init__.py

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

experiments/ml/arch.py

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

experiments/ml/evals.py

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

experiments/ml/requirements.txt

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

experiments/ml/train.py

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

experiments/procesing/pipelines.py

@@ -2,6 +2,7 @@ from sklearn.pipeline import Pipeline
 import pandas as pd
 from procesing.context import PipelineContext
 from procesing.providers import SupabaseProvider, BackendAPIProvider
+import os
 from procesing.steps import (
     FetchInteractionsStep,
     FetchPriceLogsStep,
@@ -12,11 +13,13 @@ from procesing.steps import (
     ChunkByTimeWindowStep,
     ComputeDemandForChunksStep,
     AggregatePriceLogsStep,
-    # BuildStateSpaceStep,
     FitPricingFunctionStep,
     PredictPricesStep,
     ComputeDemandStep,
-    JoinProductFeaturesStep
+    JoinProductFeaturesStep,
+    ExtractSessionFeaturesStep,
+    JoinLabelsStep,
+    ValidateDataStep,
 )
 from procesing.pricers import SimpleSurgePricer
 
@@ -106,33 +109,66 @@ def full_pipeline(context: PipelineContext,
     return product_features_df, optimal_prices_df
 
 
+def ml_training_pipeline(context: PipelineContext) -> pd.DataFrame:
+    """
+    Build labeled session-level feature matrix for ML model training.
+    Pipeline: fetch -> validate -> extract features -> join labels
+
+    Returns:
+        DataFrame with ~25 features per session + is_agent label
+        Columns: sessionId, experimentId, temporal/behavioral/product/ua features, is_agent
+    """
+    # fetch raw interactions
+    interactions_df = FetchInteractionsStep(context).transform(None)
+
+    # validate data quality (report cached in context)
+    interactions_df = ValidateDataStep(context).transform(interactions_df)
+    if interactions_df.empty:
+        return pd.DataFrame()
+
+    # extract vectorized session features
+    features_df = ExtractSessionFeaturesStep(context).transform(interactions_df)
+    if features_df.empty:
+        return pd.DataFrame()
+
+    # join experiment labels (is_agent = ~xp_human_only)
+    labeled_df = JoinLabelsStep(context).transform(features_df)
+
+    return labeled_df
+
+
 
 
 if __name__ == '__main__':
 
-    class Provider(SupabaseProvider, BackendAPIProvider):
+    class ExperimentsProvider(SupabaseProvider, BackendAPIProvider):
-        def __init__(self, backend_url: str):
-            SupabaseProvider.__init__(self)
-            BackendAPIProvider.__init__(self, backend_url=backend_url)
-
-
-    class HistoricalProvider(SupabaseProvider, BackendAPIProvider):
         def fetch_kafka_topic(self, topic: str) -> pd.DataFrame:
-            path = "/home/velocitatem/Documents/Projects/PHANTOM/experiments/collected_data/858c61ab-0a7f-4595-ae49-33f4365517b9/"
+            base_path = "/home/velocitatem/Documents/Projects/PHANTOM/experiments/collected_data/" # os.path.join(os.path.dirname(__file__), "collected_data")
-            interactions_file = "messages(2).json"
+            if not os.path.isdir(base_path):
-            prices_file = "messages(3).json"
+                return pd.DataFrame()
 
-            data = pd.read_json(path + (interactions_file if topic == "user-interactions" else prices_file))
+            files = {"user-interactions": "int.json", "price-logs": "price.json"}
-            data = [r['payload'] for r in data['value'].to_list()]
+            file_to_read = files.get(topic, files["user-interactions"])
-            data = pd.DataFrame(data)
+            frames = []
-            return data
 
+            for d in os.listdir(base_path):
+                full_path = os.path.join(base_path, d, file_to_read)
+                if not os.path.isfile(full_path):
+                    continue
+                try:
+                    data = pd.read_json(full_path)
+                    payloads = pd.DataFrame([r['payload'] for r in data['value'].to_list()])
+                    frames.append(payloads)
+                except Exception as e:
+                    print(f"Warning: Could not process {full_path}: {e}")
 
-    # example run
+            return pd.concat(frames, ignore_index=True) if frames else pd.DataFrame()
-    context = PipelineContext(
-        provider=HistoricalProvider(),
-        store_mode='airline',
-    )
 
-    product_features, prices = full_pipeline(context)
+    # demo: run ML training pipeline
-    print(prices.to_string())
+    context = PipelineContext(provider=ExperimentsProvider(), store_mode='hotel')
+    features = ml_training_pipeline(context)
+    print(f"Feature matrix: {features.shape}")
+    print(features.head())
+    print(features.info())
+
+    features.to_parquet("features.parquet")

experiments/procesing/steps/__init__.py

@@ -6,7 +6,11 @@ from procesing.steps.chunk import ChunkByTimeWindowStep
 from procesing.steps.demand import ComputeDemandStep, ComputeDemandForChunksStep
 from procesing.steps.elasticity import AggregatePriceLogsStep
 from procesing.steps.pricing import FitPricingFunctionStep, PredictPricesStep
-from procesing.steps.session import ExtractSessionFeaturesStep, _extract_features_for_session
+from procesing.steps.session import (
+    ExtractSessionFeaturesStep, JoinLabelsStep, ValidateDataStep,
+    TemporalFeatureStep, BehavioralFeatureStep, ProductFeatureStep, UserAgentFeatureStep,
+    _extract_features_for_session
+)
 
 __all__ = [
     'BaseContextStep',
@@ -25,5 +29,11 @@ __all__ = [
     'FitPricingFunctionStep',
     'PredictPricesStep',
     'ExtractSessionFeaturesStep',
+    'JoinLabelsStep',
+    'ValidateDataStep',
+    'TemporalFeatureStep',
+    'BehavioralFeatureStep',
+    'ProductFeatureStep',
+    'UserAgentFeatureStep',
     '_extract_features_for_session',
 ]

experiments/procesing/steps/base.py

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

experiments/procesing/steps/elasticity.py

@@ -7,12 +7,12 @@ class AggregatePriceLogsStep(BaseContextStep):
     """
     Aggregate price logs into time windows using VECTORIZED operations.
     Input: price_logs_df
-    Output: list of price chunks with [productId, price]
+    Output: DataFrame with columns [productId, price]
     """
 
     def transform(self, price_logs_df: pd.DataFrame):
         if price_logs_df.empty:
-            return []
+            return pd.DataFrame(columns=['productId', 'price'])
 
         df = price_logs_df.copy()
         ts_col = self.context.config.get('ts_col', 'ts')

experiments/procesing/steps/session.py

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

experiments/procesing/tests/conftest.py

@@ -269,3 +269,13 @@ def empty_context(empty_provider):
         store_mode='hotel',
         window_size='30s'
     )
+
+
+@pytest.fixture
+def session_interactions(mock_interactions):
+    """Enriched interaction data for session feature extraction tests"""
+    df = mock_interactions.copy()
+    df['userAgent'] = ['Mozilla/5.0 Chrome/120', 'Mozilla/5.0 Chrome/120',
+                       'HeadlessChrome/120', 'HeadlessChrome/120', 'HeadlessChrome/120']
+    df['metadata_base_price'] = [None, None, 150.0, 150.0, 200.0]
+    return df