#!/usr/bin/env python3 """ Sellencia POAS Calculator · v1.2 ================================ Calcula Profit-on-Ad-Spend (POAS) por SKU a partir de exports Shopify + COGS, y genera CSV listo para subir como custom_label_0 a Google Merchant Center + payload JSON para Meta CAPI value-margin. Licencia: MIT · uso libre comercial y personal Mantenido por Sellencia Growth Engineering · sellencia.com USO: python sellencia-poas-calculator.py \ --orders shopify-orders-export.csv \ --cogs cogs-per-sku.csv \ --config poas-config.json \ --out-dir ./output INPUTS: 1. shopify-orders-export.csv: export Shopify Admin → Orders → Export columnas requeridas: Name, Lineitem sku, Lineitem quantity, Lineitem price, Discount Amount, Shipping, Total, Financial Status, Fulfillment Status, Created at 2. cogs-per-sku.csv: tu coste de mercancía por SKU columnas requeridas: sku, cogs_unit columna opcional: supplier (para reporting) 3. poas-config.json: variables de negocio { "gateway_fee_pct": 0.029, // Stripe / Shopify Payments % "gateway_fee_fixed": 0.30, // fixed por transacción € "free_shipping_threshold": 60, // umbral envío gratis "shipping_cost_avg": 6.50, // coste real de envío € "returns_rate": 0.085, // 8.5% returns "returns_processing_cost": 4.20, // coste por return € "fulfillment_cost_per_order": 2.10, // pick & pack "tier_thresholds": { "high": 1.5, // POAS >= 1.5 → tier high (escalar) "med": 1.0 // POAS 1.0-1.5 → tier med (mantener) // POAS < 1.0 → tier low (pausar/optimizar) } } OUTPUTS en --out-dir: 1. poas-per-sku.csv: POAS, margen neto, tier por SKU 2. merchant-center-custom-labels.csv: listo para upload (sku, custom_label_0) 3. meta-capi-value-margin.json: SKUs con valor de margen para CAPI events 4. bidding-recommendations.md: recomendaciones de pujas por tier VALIDADO con: - Marcas DTC España €200K-€2M revenue/mes - Shopify Plus + Standard - Catálogos 50-3000 SKUs """ import argparse import csv import json import sys from collections import defaultdict from datetime import datetime from pathlib import Path # ============================================================================= # Carga de inputs # ============================================================================= def load_orders(path): """Carga export Shopify orders. Filtra a paid + fulfilled.""" orders = [] with open(path, 'r', encoding='utf-8') as f: reader = csv.DictReader(f) for row in reader: financial = (row.get('Financial Status') or '').strip().lower() fulfillment = (row.get('Fulfillment Status') or '').strip().lower() # Solo paid + fulfilled (excluye refunded, partial, pending) if financial != 'paid': continue if fulfillment not in ('fulfilled', 'partial'): continue sku = (row.get('Lineitem sku') or '').strip() if not sku: continue try: qty = int(row.get('Lineitem quantity') or 0) price = float(row.get('Lineitem price') or 0) discount = float(row.get('Discount Amount') or 0) shipping = float(row.get('Shipping') or 0) total = float(row.get('Total') or 0) except (ValueError, TypeError): continue orders.append({ 'order_name': row.get('Name', ''), 'sku': sku, 'qty': qty, 'price': price, 'discount': discount, 'shipping': shipping, 'total': total, 'created_at': row.get('Created at', ''), }) return orders def load_cogs(path): """Carga COGS por SKU.""" cogs_map = {} with open(path, 'r', encoding='utf-8') as f: reader = csv.DictReader(f) for row in reader: sku = (row.get('sku') or '').strip() if not sku: continue try: cogs_unit = float(row.get('cogs_unit') or 0) except (ValueError, TypeError): continue cogs_map[sku] = { 'cogs_unit': cogs_unit, 'supplier': row.get('supplier', ''), } return cogs_map def load_config(path): """Carga config JSON con variables de negocio.""" with open(path, 'r', encoding='utf-8') as f: return json.load(f) # ============================================================================= # Cálculo POAS # ============================================================================= def aggregate_by_sku(orders): """Agrega métricas por SKU.""" agg = defaultdict(lambda: { 'units_sold': 0, 'gross_revenue': 0.0, 'discount_total': 0.0, 'shipping_collected': 0.0, 'orders_count': 0, }) seen_orders = defaultdict(set) for o in orders: sku = o['sku'] agg[sku]['units_sold'] += o['qty'] agg[sku]['gross_revenue'] += o['price'] * o['qty'] agg[sku]['discount_total'] += o['discount'] # Shipping y order count se cuentan una vez por order if o['order_name'] not in seen_orders[sku]: agg[sku]['shipping_collected'] += o['shipping'] agg[sku]['orders_count'] += 1 seen_orders[sku].add(o['order_name']) return agg def calculate_poas(sku_agg, cogs_map, config): """Calcula margen neto y POAS por SKU. POAS = Net Margin / Ad Spend Aquí calculamos solo el margin side. Ad Spend se inyecta luego desde Google Ads / Meta Ads exports (ver merge_with_ad_spend()). """ results = {} gateway_pct = config.get('gateway_fee_pct', 0.029) gateway_fixed = config.get('gateway_fee_fixed', 0.30) free_ship_threshold = config.get('free_shipping_threshold', 60) shipping_cost = config.get('shipping_cost_avg', 6.50) returns_rate = config.get('returns_rate', 0.085) returns_cost = config.get('returns_processing_cost', 4.20) fulfillment_cost = config.get('fulfillment_cost_per_order', 2.10) for sku, m in sku_agg.items(): cogs_data = cogs_map.get(sku) if not cogs_data: # SKU sin COGS configurado — flag para review manual results[sku] = { 'units_sold': m['units_sold'], 'gross_revenue': round(m['gross_revenue'], 2), 'cogs_total': None, 'net_margin': None, 'margin_pct': None, 'tier': 'unknown', 'flag': 'cogs_missing', } continue cogs_unit = cogs_data['cogs_unit'] units = m['units_sold'] orders_count = m['orders_count'] gross = m['gross_revenue'] discount = m['discount_total'] shipping_collected = m['shipping_collected'] # COGS total cogs_total = cogs_unit * units # Net revenue (gross - discounts) net_revenue = gross - discount # Gateway fees: % sobre net revenue + fixed por order gateway_total = (net_revenue * gateway_pct) + (gateway_fixed * orders_count) # Shipping cost real menos lo cobrado al cliente shipping_total_cost = shipping_cost * orders_count shipping_net_cost = shipping_total_cost - shipping_collected # Fulfillment (pick & pack, materiales) fulfillment_total = fulfillment_cost * orders_count # Returns expected: % del net_revenue se devuelve returns_revenue_lost = net_revenue * returns_rate returns_processing = returns_cost * (orders_count * returns_rate) # NET MARGIN final (excluye ad spend, eso se cruza luego) net_margin = ( net_revenue - cogs_total - gateway_total - shipping_net_cost - fulfillment_total - returns_revenue_lost - returns_processing ) margin_pct = (net_margin / net_revenue) if net_revenue > 0 else 0 results[sku] = { 'units_sold': units, 'orders_count': orders_count, 'gross_revenue': round(gross, 2), 'net_revenue': round(net_revenue, 2), 'cogs_total': round(cogs_total, 2), 'gateway_fees': round(gateway_total, 2), 'shipping_net_cost': round(shipping_net_cost, 2), 'fulfillment_cost': round(fulfillment_total, 2), 'returns_impact': round(returns_revenue_lost + returns_processing, 2), 'net_margin': round(net_margin, 2), 'margin_pct': round(margin_pct * 100, 2), 'margin_per_unit': round(net_margin / units, 2) if units > 0 else 0, 'tier': classify_tier_from_margin(margin_pct, config), 'flag': 'ok' if margin_pct > 0 else 'negative_margin', } return results def classify_tier_from_margin(margin_pct, config): """Clasifica tier por margin% si no hay ad spend cruzado todavía. Heurística inicial: - margin_pct >= 35% → high - margin_pct 20-35% → med - margin_pct 10-20% → low - margin_pct < 10% → toxic """ if margin_pct >= 0.35: return 'high' elif margin_pct >= 0.20: return 'med' elif margin_pct >= 0.10: return 'low' else: return 'toxic' def reclassify_with_poas(results, ad_spend_map, config): """Re-clasifica tier usando POAS real cuando hay ad_spend disponible. POAS = net_margin / ad_spend_attributed_to_sku ad_spend_map: dict {sku: ad_spend_eur} desde Google Ads + Meta Ads exports """ high_thr = config.get('tier_thresholds', {}).get('high', 1.5) med_thr = config.get('tier_thresholds', {}).get('med', 1.0) for sku, data in results.items(): ad_spend = ad_spend_map.get(sku, 0) net_margin = data.get('net_margin') if net_margin is None or ad_spend == 0: data['poas'] = None data['ad_spend'] = ad_spend continue poas = net_margin / ad_spend data['poas'] = round(poas, 2) data['ad_spend'] = round(ad_spend, 2) if poas >= high_thr: data['tier'] = 'high' elif poas >= med_thr: data['tier'] = 'med' elif poas > 0: data['tier'] = 'low' else: data['tier'] = 'toxic' return results # ============================================================================= # Outputs # ============================================================================= def write_poas_csv(results, out_path): """CSV completo con todas las métricas POAS por SKU.""" fieldnames = [ 'sku', 'units_sold', 'orders_count', 'gross_revenue', 'net_revenue', 'cogs_total', 'gateway_fees', 'shipping_net_cost', 'fulfillment_cost', 'returns_impact', 'net_margin', 'margin_pct', 'margin_per_unit', 'ad_spend', 'poas', 'tier', 'flag', ] with open(out_path, 'w', newline='', encoding='utf-8') as f: writer = csv.DictWriter(f, fieldnames=fieldnames) writer.writeheader() for sku, data in sorted(results.items(), key=lambda x: -(x[1].get('net_margin') or 0)): row = {'sku': sku} for k in fieldnames[1:]: row[k] = data.get(k, '') writer.writerow(row) def write_merchant_center_csv(results, out_path): """CSV listo para upload a Google Merchant Center con custom_label_0 = tier.""" with open(out_path, 'w', newline='', encoding='utf-8') as f: writer = csv.writer(f) writer.writerow(['id', 'custom_label_0']) for sku, data in results.items(): tier = data.get('tier', 'unknown') # Skip toxic y unknown — no querrás push agresivo en estos if tier in ('toxic', 'unknown'): continue writer.writerow([sku, f'poas_{tier}']) def write_meta_capi_json(results, out_path): """JSON con SKU → margen para inyectar en Meta CAPI value-margin events. Úsalo en tu Cloudflare Worker / sGTM para enviar el campo 'value' al Meta CAPI con el margen real, no el revenue. """ capi_data = { 'generated_at': datetime.utcnow().isoformat() + 'Z', 'currency': 'EUR', 'value_strategy': 'net_margin', 'skus': {} } for sku, data in results.items(): if data.get('flag') == 'cogs_missing': continue capi_data['skus'][sku] = { 'margin_per_unit': data.get('margin_per_unit', 0), 'tier': data.get('tier', 'unknown'), 'send_value_as_margin': data.get('margin_per_unit', 0) > 0, } with open(out_path, 'w', encoding='utf-8') as f: json.dump(capi_data, f, indent=2, ensure_ascii=False) def write_bidding_recommendations(results, out_path): """Markdown con recomendaciones de pujas por tier.""" tier_buckets = defaultdict(list) for sku, data in results.items(): tier_buckets[data.get('tier', 'unknown')].append((sku, data)) lines = [ '# Recomendaciones de pujas por tier · POAS framework', f'Generado: {datetime.utcnow().strftime("%Y-%m-%d %H:%M UTC")}', '', '## Distribución por tier', '', ] for tier in ['high', 'med', 'low', 'toxic', 'unknown']: bucket = tier_buckets.get(tier, []) total_margin = sum((d.get('net_margin') or 0) for _, d in bucket) lines.append(f'- **{tier}**: {len(bucket)} SKUs · margen agregado €{total_margin:,.0f}') lines.extend([ '', '## Acciones recomendadas', '', '### Tier `high` (POAS ≥ 1.5)', '- Subir target ROAS Google Ads PMax: -10% (más agresivo)', '- Aumentar bid cap Meta Ads ASC: +15-20%', '- Custom label `poas_high` para campaña dedicada', '- Free shipping siempre activo en estos SKUs', '', '### Tier `med` (POAS 1.0-1.5)', '- Mantener configuración actual', '- Custom label `poas_med`', '- Test creativos para mover a `high`', '', '### Tier `low` (POAS 0-1.0)', '- Bajar target ROAS Google Ads: +25% (más conservador)', '- Reducir bid cap Meta Ads: -20%', '- Custom label `poas_low`', '- Plantear bundle con SKU `high` para subir AOV', '', '### Tier `toxic` (POAS negativo)', '- Pausar campañas dedicadas', '- Excluir de PMax feed via custom_label si no son loss-leaders estratégicos', '- Revisar pricing base o COGS supplier', '', '## Top 10 SKUs por margen absoluto', '', '| SKU | Net margin | POAS | Tier |', '|---|---|---|---|', ]) sorted_results = sorted( results.items(), key=lambda x: -(x[1].get('net_margin') or 0) )[:10] for sku, data in sorted_results: nm = data.get('net_margin') or 0 poas = data.get('poas') poas_str = f'{poas:.2f}' if poas is not None else 'n/a' tier = data.get('tier', '-') lines.append(f'| {sku} | €{nm:,.0f} | {poas_str} | {tier} |') lines.extend([ '', '## Top 10 SKUs `toxic` (revisar urgente)', '', '| SKU | Net margin | Margin % | Flag |', '|---|---|---|---|', ]) toxic = [ (sku, d) for sku, d in results.items() if d.get('tier') == 'toxic' and d.get('net_margin') is not None ] toxic_sorted = sorted(toxic, key=lambda x: x[1].get('net_margin') or 0)[:10] for sku, data in toxic_sorted: nm = data.get('net_margin') or 0 mp = data.get('margin_pct') or 0 flag = data.get('flag', '-') lines.append(f'| {sku} | €{nm:,.0f} | {mp:.1f}% | {flag} |') lines.append('') lines.append('---') lines.append('Sellencia Growth Engineering · sellencia.com') with open(out_path, 'w', encoding='utf-8') as f: f.write('\n'.join(lines)) # ============================================================================= # Ad spend optional input # ============================================================================= def load_ad_spend(path): """Opcional. CSV con sku,ad_spend_eur agregado de Google Ads + Meta Ads. Si no se proporciona, tier se clasifica por margin_pct (heurística inicial). """ if not path: return {} ad_map = {} with open(path, 'r', encoding='utf-8') as f: reader = csv.DictReader(f) for row in reader: sku = (row.get('sku') or '').strip() if not sku: continue try: spend = float(row.get('ad_spend_eur') or 0) except (ValueError, TypeError): continue ad_map[sku] = spend return ad_map # ============================================================================= # CLI # ============================================================================= def main(): parser = argparse.ArgumentParser( description='Sellencia POAS Calculator · DTC ecommerce', formatter_class=argparse.RawDescriptionHelpFormatter, epilog=__doc__, ) parser.add_argument('--orders', required=True, help='Shopify orders CSV') parser.add_argument('--cogs', required=True, help='COGS per SKU CSV') parser.add_argument('--config', required=True, help='Business config JSON') parser.add_argument('--ad-spend', help='Optional: ad spend per SKU CSV') parser.add_argument('--out-dir', default='./output', help='Output directory') args = parser.parse_args() out_dir = Path(args.out_dir) out_dir.mkdir(parents=True, exist_ok=True) print(f'[+] Loading orders from {args.orders}') orders = load_orders(args.orders) print(f' Loaded {len(orders)} line items (paid + fulfilled only)') print(f'[+] Loading COGS from {args.cogs}') cogs_map = load_cogs(args.cogs) print(f' Loaded {len(cogs_map)} SKUs with COGS') print(f'[+] Loading config from {args.config}') config = load_config(args.config) print('[+] Aggregating by SKU') sku_agg = aggregate_by_sku(orders) print(f' {len(sku_agg)} unique SKUs sold') print('[+] Calculating POAS') results = calculate_poas(sku_agg, cogs_map, config) if args.ad_spend: print(f'[+] Loading ad spend from {args.ad_spend}') ad_spend_map = load_ad_spend(args.ad_spend) print(f' Loaded ad spend for {len(ad_spend_map)} SKUs') results = reclassify_with_poas(results, ad_spend_map, config) print('[+] Writing outputs') write_poas_csv(results, out_dir / 'poas-per-sku.csv') write_merchant_center_csv(results, out_dir / 'merchant-center-custom-labels.csv') write_meta_capi_json(results, out_dir / 'meta-capi-value-margin.json') write_bidding_recommendations(results, out_dir / 'bidding-recommendations.md') # Summary total_skus = len(results) by_tier = defaultdict(int) for data in results.values(): by_tier[data.get('tier', 'unknown')] += 1 print('') print('=' * 60) print('SUMMARY') print('=' * 60) print(f'Total SKUs analyzed: {total_skus}') for tier in ['high', 'med', 'low', 'toxic', 'unknown']: count = by_tier.get(tier, 0) pct = (count / total_skus * 100) if total_skus else 0 print(f' {tier:10s}: {count:5d} SKUs ({pct:.1f}%)') total_margin = sum((d.get('net_margin') or 0) for d in results.values()) print(f'Net margin agregado: €{total_margin:,.2f}') if args.ad_spend: total_ad = sum((d.get('ad_spend') or 0) for d in results.values()) if total_ad > 0: blended_poas = total_margin / total_ad print(f'Ad spend agregado: €{total_ad:,.2f}') print(f'POAS blended: {blended_poas:.2f}') print('') print(f'Outputs en: {out_dir.resolve()}') print(' - poas-per-sku.csv') print(' - merchant-center-custom-labels.csv') print(' - meta-capi-value-margin.json') print(' - bidding-recommendations.md') if __name__ == '__main__': sys.exit(main())