import math import logging import re from typing import Optional from aura_core.module_contract import aura_module, ModuleResult from aura_core.types import ( ValuationInput, DomainIntrinsic, TrafficEstimate, MonthlyRevenue, RevenueModel, NicheValuation, ValuationResult, ) logger = logging.getLogger(__name__) BUSINESS_MODELS = { "affiliate": { "label": "Affiliate / Referral", "fulfillment_cost_pct": 0.0, "startup_cost_low": 200, "startup_cost_high": 2000, "margin_pct": 0.85, "monthly_overhead_low": 20, "monthly_overhead_high": 100, "time_to_first_dollar_days": 60, "scalability": 0.95, "risk_factor": 0.25, "description": "Earn commissions referring visitors to partner products. Near-zero fulfillment — you never touch product, handle returns, or manage inventory.", }, "digital": { "label": "Digital Products / Info", "fulfillment_cost_pct": 0.03, "startup_cost_low": 500, "startup_cost_high": 5000, "margin_pct": 0.90, "monthly_overhead_low": 30, "monthly_overhead_high": 150, "time_to_first_dollar_days": 45, "scalability": 0.92, "risk_factor": 0.20, "description": "Sell courses, templates, ebooks, tools, or SaaS subscriptions. Fulfillment is automated — once built, each sale costs virtually nothing to deliver.", }, "saas": { "label": "SaaS / Subscription", "fulfillment_cost_pct": 0.10, "startup_cost_low": 5000, "startup_cost_high": 50000, "margin_pct": 0.75, "monthly_overhead_low": 200, "monthly_overhead_high": 2000, "time_to_first_dollar_days": 120, "scalability": 0.90, "risk_factor": 0.35, "description": "Recurring subscription revenue. Fulfillment means server costs and support — scales well but needs ongoing development and customer success.", }, "ecommerce_dropship": { "label": "E-commerce (Dropship)", "fulfillment_cost_pct": 0.35, "startup_cost_low": 1000, "startup_cost_high": 8000, "margin_pct": 0.25, "monthly_overhead_low": 100, "monthly_overhead_high": 500, "time_to_first_dollar_days": 30, "scalability": 0.60, "risk_factor": 0.40, "description": "Sell physical products without holding inventory. Supplier ships direct, but you handle customer service, returns, and quality control. Margins are tight.", }, "ecommerce_inventory": { "label": "E-commerce (Own Inventory)", "fulfillment_cost_pct": 0.50, "startup_cost_low": 5000, "startup_cost_high": 50000, "margin_pct": 0.35, "monthly_overhead_low": 500, "monthly_overhead_high": 5000, "time_to_first_dollar_days": 60, "scalability": 0.40, "risk_factor": 0.55, "description": "Buy, store, pack, and ship physical products yourself. Highest fulfillment burden — warehousing, shipping labels, returns, damaged goods, and inventory risk.", }, "services": { "label": "Service Business", "fulfillment_cost_pct": 0.45, "startup_cost_low": 500, "startup_cost_high": 10000, "margin_pct": 0.50, "monthly_overhead_low": 100, "monthly_overhead_high": 1000, "time_to_first_dollar_days": 30, "scalability": 0.30, "risk_factor": 0.30, "description": "Trade time for money — consulting, freelancing, agency work. Fulfillment is your labor or hiring others. Hard to scale without hiring.", }, } TLD_MULTIPLIERS = { ".com": 1.0, ".net": 0.55, ".org": 0.50, ".io": 0.70, ".co": 0.60, ".ai": 0.80, ".app": 0.50, ".dev": 0.45, ".biz": 0.30, ".us": 0.35, ".info": 0.25, ".xyz": 0.20, ".online": 0.20, ".store": 0.30, ".shop": 0.30, ".tech": 0.40, ".me": 0.40, ".tv": 0.50, } def _parse_valuation_band(band_str) -> tuple[int, int]: if not band_str: return 5000, 20000 band_str = str(band_str).replace(",", "").replace("$", "").strip() match = re.findall(r"(\d+)", band_str) if len(match) >= 2: return int(match[0]), int(match[1]) elif len(match) == 1: v = int(match[0]) return v, v * 3 return 5000, 20000 def compute_domain_intrinsic(domain: str) -> DomainIntrinsic: parts = domain.rsplit(".", 1) name = parts[0] if parts else domain tld = "." + parts[1] if len(parts) > 1 else ".com" length = len(name.replace("-", "")) tld_mult = TLD_MULTIPLIERS.get(tld, 0.25) if length <= 3: length_score = 1.0 elif length <= 5: length_score = 0.85 elif length <= 8: length_score = 0.65 elif length <= 12: length_score = 0.45 elif length <= 16: length_score = 0.30 else: length_score = 0.15 has_hyphens = "-" in name hyphen_penalty = 0.6 if has_hyphens else 1.0 has_numbers = any(c.isdigit() for c in name) number_penalty = 0.75 if has_numbers else 1.0 is_dictionary_like = not has_hyphens and not has_numbers and length <= 10 dictionary_bonus = 1.3 if is_dictionary_like else 1.0 raw_score = length_score * tld_mult * hyphen_penalty * number_penalty * dictionary_bonus raw_score = min(1.0, raw_score) base_floor = 500 base_ceiling = 150000 domain_only_value = base_floor + (base_ceiling - base_floor) * (raw_score ** 2.0) return DomainIntrinsic( domain_name=domain, name_part=name, tld=tld, length=length, tld_multiplier=tld_mult, length_score=round(length_score, 2), has_hyphens=has_hyphens, has_numbers=has_numbers, raw_score=round(raw_score, 3), domain_only_value=round(domain_only_value), ) def classify_monetization(monetization_str: Optional[str]) -> str: if not monetization_str: return "affiliate" m = monetization_str.lower() if "saas" in m or "subscription" in m: return "saas" if "affiliate" in m or "referral" in m or "commission" in m: return "affiliate" if "digital" in m or "info" in m or "course" in m or "ebook" in m or "template" in m or "product" in m: return "digital" if "e-commerce" in m or "ecommerce" in m or "shop" in m or "store" in m: return "ecommerce_dropship" if "service" in m or "consult" in m or "agency" in m or "freelance" in m: return "services" return "affiliate" def estimate_monthly_traffic(niche_data: dict, domain_score: DomainIntrinsic) -> TrafficEstimate: audience = str(niche_data.get("target_audience", "")) audience_segments = [s.strip() for s in re.split(r"[,;&]", audience) if s.strip()] num_segments = max(1, len(audience_segments)) affiliates = niche_data.get("affiliate_programs", []) num_affiliates = len(affiliates) if isinstance(affiliates, list) else 0 time_to_rev = niche_data.get("time_to_revenue", "medium") if time_to_rev == "fast": market_heat = 1.3 elif time_to_rev == "slow": market_heat = 0.7 else: market_heat = 1.0 tld_trust = domain_score.tld_multiplier length_factor = domain_score.length_score base_monthly_visitors = 800 segment_mult = min(3.0, 1.0 + (num_segments - 1) * 0.25) affiliate_mult = min(2.0, 1.0 + num_affiliates * 0.1) monthly_low = base_monthly_visitors * segment_mult * market_heat * tld_trust * length_factor * 0.6 monthly_high = base_monthly_visitors * segment_mult * affiliate_mult * market_heat * tld_trust * length_factor * 2.5 monthly_low = max(200, round(monthly_low)) monthly_high = max(monthly_low * 2, round(monthly_high)) return TrafficEstimate( audience_segments=num_segments, affiliate_count=num_affiliates, market_heat=market_heat, monthly_visitors_low=monthly_low, monthly_visitors_high=monthly_high, monthly_visitors_mid=(monthly_low + monthly_high) // 2, ) def compute_revenue_by_model(traffic: TrafficEstimate, niche_data: dict) -> list[RevenueModel]: results = [] primary_model = classify_monetization(niche_data.get("monetization_model", "")) requires_inventory = niche_data.get("requires_inventory", False) models_to_eval = ["affiliate", "digital"] if primary_model == "saas": models_to_eval.append("saas") if requires_inventory: models_to_eval.append("ecommerce_inventory") elif primary_model in ("ecommerce_dropship", "ecommerce_inventory"): models_to_eval.append("ecommerce_dropship") if primary_model not in models_to_eval: models_to_eval.append(primary_model) visitors_low = traffic.monthly_visitors_low visitors_high = traffic.monthly_visitors_high for model_key in models_to_eval: model = BUSINESS_MODELS[model_key] is_primary = model_key == primary_model if model_key == "affiliate": conv_rate = 0.02 avg_commission = 25 revenue_per_visitor = conv_rate * avg_commission elif model_key == "digital": conv_rate = 0.015 avg_price = 47 revenue_per_visitor = conv_rate * avg_price elif model_key == "saas": conv_rate = 0.008 avg_mrr = 29 revenue_per_visitor = conv_rate * avg_mrr elif model_key == "ecommerce_dropship": conv_rate = 0.02 avg_order = 65 revenue_per_visitor = conv_rate * avg_order * model["margin_pct"] elif model_key == "ecommerce_inventory": conv_rate = 0.025 avg_order = 75 revenue_per_visitor = conv_rate * avg_order * model["margin_pct"] elif model_key == "services": conv_rate = 0.005 avg_engagement = 500 revenue_per_visitor = conv_rate * avg_engagement else: revenue_per_visitor = 0.50 gross_low = visitors_low * revenue_per_visitor gross_high = visitors_high * revenue_per_visitor fulfillment_low = gross_low * model["fulfillment_cost_pct"] fulfillment_high = gross_high * model["fulfillment_cost_pct"] net_low = gross_low - fulfillment_low - model["monthly_overhead_low"] net_high = gross_high - fulfillment_high - model["monthly_overhead_high"] avg_gross = (gross_low + gross_high) / 2 avg_net = (net_low + net_high) / 2 profit_margin_pct = round(avg_net / max(1, avg_gross) * 100) annual_net_low = net_low * 12 annual_net_high = net_high * 12 if model_key in ("affiliate", "digital"): valuation_mult = 36 elif model_key == "saas": valuation_mult = 48 elif model_key in ("ecommerce_dropship", "ecommerce_inventory"): valuation_mult = 24 else: valuation_mult = 18 developed_value_low = round(max(0, annual_net_low) / 12 * valuation_mult) developed_value_high = round(max(0, annual_net_high) / 12 * valuation_mult) results.append(RevenueModel( model_key=model_key, label=model["label"], is_primary_fit=is_primary, description=model["description"], fulfillment_cost_pct=round(model["fulfillment_cost_pct"] * 100), startup_cost={"low": model["startup_cost_low"], "high": model["startup_cost_high"]}, time_to_first_dollar_days=model["time_to_first_dollar_days"], scalability_score=round(model["scalability"] * 10, 1), risk_score=round(model["risk_factor"] * 10, 1), monthly_revenue=MonthlyRevenue( gross_low=round(gross_low), gross_high=round(gross_high), fulfillment_cost_low=round(fulfillment_low), fulfillment_cost_high=round(fulfillment_high), overhead_low=model["monthly_overhead_low"], overhead_high=model["monthly_overhead_high"], net_profit_low=round(net_low), net_profit_high=round(net_high), profit_margin_pct=profit_margin_pct, ), annual_net={"low": round(annual_net_low), "high": round(annual_net_high)}, developed_value={"low": developed_value_low, "high": developed_value_high}, valuation_multiple=f"{valuation_mult}x monthly net", )) results.sort(key=lambda x: (not x.is_primary_fit, -x.monthly_revenue.net_profit_high)) return results def valuate_domain(domain: str, analysis: dict) -> ValuationResult: domain_score = compute_domain_intrinsic(domain) niches = analysis.get("niches", []) niche_valuations = [] best_developed_high = 0 best_monthly_high = 0 for niche in niches: val_low, val_high = _parse_valuation_band(niche.get("valuation_band")) traffic = estimate_monthly_traffic(niche, domain_score) revenue_models = compute_revenue_by_model(traffic, niche) best_model = revenue_models[0] if revenue_models else None niche_developed_high = max((m.developed_value["high"] for m in revenue_models), default=0) niche_monthly_high = max((m.monthly_revenue.net_profit_high for m in revenue_models), default=0) if niche_developed_high > best_developed_high: best_developed_high = niche_developed_high if niche_monthly_high > best_monthly_high: best_monthly_high = niche_monthly_high niche_valuations.append(NicheValuation( niche_name=niche.get("name", "Unknown"), monetization_model=niche.get("monetization_model", ""), requires_inventory=niche.get("requires_inventory", False), ai_valuation_band={"low": val_low, "high": val_high}, traffic_estimate=traffic, revenue_by_model=revenue_models, best_model={ "label": best_model.label if best_model else "N/A", "net_monthly_high": best_model.monthly_revenue.net_profit_high if best_model else 0, "developed_value_high": best_model.developed_value["high"] if best_model else 0, } if best_model else None, )) fulfillment_summary = { "zero_fulfillment": ["Affiliate / Referral"], "near_zero": ["Digital Products / Info"], "moderate": ["SaaS / Subscription"], "heavy": ["E-commerce (Dropship)", "Service Business"], "maximum": ["E-commerce (Own Inventory)"], } return ValuationResult( domain=domain, domain_intrinsic=domain_score, domain_only_value=domain_score.domain_only_value, best_developed_value=best_developed_high, best_monthly_net=best_monthly_high, niches_evaluated=len(niche_valuations), niche_valuations=niche_valuations, fulfillment_guide=fulfillment_summary, business_models={ k: { "label": v["label"], "fulfillment_cost_pct": round(v["fulfillment_cost_pct"] * 100), "margin_pct": round(v["margin_pct"] * 100), "description": v["description"], } for k, v in BUSINESS_MODELS.items() }, ) @aura_module( name="valuation", version="1.0.0", description="Domain valuation engine — computes intrinsic domain value, traffic estimates, revenue projections across 6 business models, and developed-value ranges", input_model=ValuationInput, tags=["valuation", "domain", "revenue"], ) def run_valuation(config: dict) -> ModuleResult: domain = config.get("domain", "example.com") analysis = config.get("analysis", {}) result = valuate_domain(domain, analysis) output = result.model_dump() if hasattr(result, "model_dump") else result.__dict__ return ModuleResult( ok=True, output=output, metadata={"domain": domain, "niches_evaluated": result.niches_evaluated}, )