Join programmatic ad data with retail & inventory systems to calculate total ROAS across marketing channels
triggers
["combine ad spend with sales data","join marketing data with inventory","calculate cross-channel ROAS","merge programmatic ads and retail data","integrate advertising and sales metrics","unify marketing and retail attribution","connect ad platforms with point of sale","aggregate multi-channel marketing performance"]
The GenPark Cross-Channel Marketing Data Joiner is a Python skill that unifies programmatic advertising data with retail and inventory systems to provide comprehensive ROAS (Return on Ad Spend) analytics. It bridges the gap between digital marketing campaigns and physical/online sales outcomes, enabling marketers to understand true campaign effectiveness across channels.
Installation
# Clone the repository
git clone https://github.com/alphaparkinc/genpark-cross-channel-marketing-data-joiner-skill.git
cd genpark-cross-channel-marketing-data-joiner-skill
# Install dependencies
pip install -r requirements.txt
from genpark_data_joiner import load_config
config = load_config("config.yaml")
joiner = CrossChannelJoiner.from_config(config)
Environment Variables
# Ad Platform Credentialsexport GOOGLE_ADS_API_KEY="your_key"export GOOGLE_ADS_CUSTOMER_ID="your_customer_id"export FB_ADS_API_KEY="your_key"export FB_AD_ACCOUNT_ID="your_account_id"export DV360_CREDENTIALS="path/to/credentials.json"# Retail Platform Credentialsexport SHOPIFY_API_KEY="your_key"export SHOPIFY_STORE_URL="your-store.myshopify.com"export SQUARE_ACCESS_TOKEN="your_token"# Inventory System Credentialsexport WAREHOUSE_ENDPOINT="https://api.warehouse.example.com"export WAREHOUSE_AUTH_TOKEN="your_token"
Common Patterns
Pattern 1: Weekly ROAS Report
from genpark_data_joiner import CrossChannelJoiner, ReportGenerator
from datetime import datetime, timedelta
defgenerate_weekly_roas_report():
# Get last 7 days
end_date = datetime.now()
start_date = end_date - timedelta(days=7)
# Initialize and join data
joiner = CrossChannelJoiner.from_config("config.yaml")
result = joiner.join_and_calculate(
date_range=(start_date, end_date),
attribution_window=7
)
# Generate report
report = ReportGenerator(result)
report.add_summary()
report.add_channel_breakdown()
report.add_top_campaigns(limit=10)
report.add_product_performance()
# Export
report.export_csv("weekly_roas_report.csv")
report.export_pdf("weekly_roas_report.pdf")
return result
if __name__ == "__main__":
generate_weekly_roas_report()
Pattern 2: Multi-Attribution Comparison
from genpark_data_joiner import CrossChannelJoiner, AttributionModel
defcompare_attribution_models(start_date, end_date):
joiner = CrossChannelJoiner.from_config("config.yaml")
models = [
AttributionModel.LAST_CLICK,
AttributionModel.FIRST_CLICK,
AttributionModel.LINEAR,
AttributionModel.TIME_DECAY,
AttributionModel.POSITION_BASED
]
results = {}
for model in models:
result = joiner.join_and_calculate(
date_range=(start_date, end_date),
attribution_model=model,
attribution_window=14
)
results[model.name] = {
"total_roas": result.total_roas,
"channel_roas": result.channel_roas
}
# Compare resultsfor model_name, data in results.items():
print(f"\n{model_name}:")
print(f" Total ROAS: {data['total_roas']:.2f}")
for channel, roas in data['channel_roas'].items():
print(f" {channel}: {roas:.2f}")
return results
Pattern 3: Real-Time ROAS Dashboard
from genpark_data_joiner import CrossChannelJoiner, StreamingConnector
import time
defrealtime_roas_monitor(refresh_interval=300):
"""Monitor ROAS every 5 minutes"""
joiner = CrossChannelJoiner.from_config("config.yaml")
whileTrue:
try:
# Get today's data
result = joiner.join_and_calculate(
date_range="today",
attribution_window=1
)
print(f"\n[{datetime.now()}] Real-time ROAS:")
print(f"Total ROAS: {result.total_roas:.2f}")
print(f"Total Spend: ${result.total_spend:,.2f}")
print(f"Total Revenue: ${result.total_revenue:,.2f}")
# Alert if ROAS drops below thresholdif result.total_roas < 2.0:
send_alert(f"ROAS Alert: {result.total_roas:.2f}")
time.sleep(refresh_interval)
except Exception as e:
print(f"Error: {e}")
time.sleep(60)
Pattern 4: Inventory-Aware Campaign Optimization
from genpark_data_joiner import CrossChannelJoiner, InventoryOptimizer
defoptimize_campaigns_by_inventory():
joiner = CrossChannelJoiner.from_config("config.yaml")
optimizer = InventoryOptimizer(joiner)
# Get current campaign performance with inventory levels
analysis = optimizer.analyze(
include_stock_levels=True,
include_margins=True,
include_velocity=True
)
recommendations = []
for campaign in analysis.campaigns:
if campaign.inventory_level == "low"and campaign.roas > 3.0:
recommendations.append({
"campaign_id": campaign.id,
"action": "pause",
"reason": "Low inventory, high ROAS - avoid stockout"
})
elif campaign.inventory_level == "high"and campaign.roas < 1.5:
recommendations.append({
"campaign_id": campaign.id,
"action": "increase_budget",
"reason": "High inventory, low ROAS - clear stock"
})
return recommendations
Troubleshooting
Connection Issues
# Test individual connectionsfrom genpark_data_joiner import test_connections
results = test_connections("config.yaml")
for source, status in results.items():
ifnot status["connected"]:
print(f"Failed to connect to {source}: {status['error']}")
