mirror of
https://github.com/velocitatem/PHANTOM.git
synced 2026-05-31 08:33:36 +00:00
8751583764
* fix: fixes of backwords * fixing hotel information with image placeholders * chore: clean up product display in hotel and cleaner interfacing * adding loader with historical data loading * feature: cleaning up pipeline * chore: simple surge pricer * created new pricing pipeline * adding a checkout page to both sites * fix: fixing stale pacakge * test: we wont be using elasticity anymore so its okay * chore: cleaning elasticity references * chore: store sting * feature: e2e intro pipline surge pricing * fix: CVE vulnerability patching
59 lines
2.1 KiB
Python
Executable File
59 lines
2.1 KiB
Python
Executable File
import pandas as pd
|
|
from procesing.steps.base import BaseContextStep
|
|
|
|
class JoinExperimentsStep(BaseContextStep):
|
|
"""Join experiment metadata to interactions"""
|
|
|
|
def transform(self, data: tuple):
|
|
"""
|
|
Args:
|
|
data: (interactions_df, experiments_df)
|
|
Returns:
|
|
merged interactions dataframe
|
|
"""
|
|
interactions_df, experiments_df = data
|
|
|
|
if experiments_df.empty:
|
|
return interactions_df
|
|
|
|
# Flatten nested task field if present
|
|
if 'task' in experiments_df.columns and experiments_df['task'].notnull().any():
|
|
task_norm = pd.json_normalize(experiments_df['task'].dropna())
|
|
task_norm.index = experiments_df[experiments_df['task'].notnull()].index
|
|
experiments_df = experiments_df.drop('task', axis=1).join(task_norm, rsuffix='_task')
|
|
|
|
# Rename for clarity
|
|
experiments_df = experiments_df.rename(columns={
|
|
'id': 'experimentId',
|
|
'subject_name': 'exp_subject',
|
|
'xp_human_only': 'exp_human_only',
|
|
'xp_market_mode': 'exp_market_mode',
|
|
'xp_task_id': 'exp_task_id'
|
|
})
|
|
|
|
return interactions_df.merge(experiments_df, on='experimentId', how='left')
|
|
|
|
class JoinProductFeaturesStep(BaseContextStep):
|
|
"""Join product features to interactions"""
|
|
|
|
def transform(self, data: tuple):
|
|
"""
|
|
Args:
|
|
data: (interactions_df, products_df)
|
|
Returns:
|
|
merged interactions dataframe
|
|
"""
|
|
demand_df, price_df = data
|
|
|
|
# get base prices from products if available
|
|
products = self.context.products
|
|
products['base_price'] = products.apply(
|
|
lambda row: float(row['metadata'].get('base_price', 0.0)) if isinstance(row['metadata'], dict) else 0,
|
|
axis=1
|
|
)
|
|
products = products[['id', 'base_price']].rename(columns={'id': 'productId'})
|
|
|
|
if price_df.empty:
|
|
return demand_df
|
|
return demand_df.merge(price_df, on='productId', how='left').merge(products, on='productId', how='left')
|