"""Aswath Damodaran's published datasets, as data inputs to the valuation. Source: https://pages.stern.nyu.edu/~adamodar/ (datasets are his; he updates the industry files each January, the country risk premiums in January and July, and the implied equity risk premium monthly). Nothing here is fitted to his valuations: these are the same inputs he uses. * ``build(src_dir)`` parses the downloaded files into ``damodaran_data.json.gz`` (run locally or by the refresh chain; needs xlrd/openpyxl). The Space only loads the JSON. * Loader functions: ``implied_erp``, ``country_premium``, ``company_industry``, ``industry_stats``, ``bottom_up_beta``, ``company_erp``. """ from __future__ import annotations import datetime as _dt import gzip import json import os import re DATA_FILE = os.path.join(os.path.dirname(os.path.abspath(__file__)), "damodaran_data.json.gz") BASE_URL = "https://pages.stern.nyu.edu/~adamodar/pc/" SOURCES = { "betas_us": "datasets/betas.xls", "betas_global": "datasets/betaGlobal.xls", "capex_us": "datasets/capex.xls", "capex_global": "datasets/capexGlobal.xls", "margin_us": "datasets/margin.xls", "margin_global": "datasets/marginGlobal.xls", "wacc_us": "datasets/wacc.xls", "wacc_global": "datasets/waccGlobal.xls", "ps_us": "datasets/psdata.xls", "ps_global": "datasets/psGlobal.xls", "pe_us": "datasets/pedata.xls", "pe_global": "datasets/peGlobal.xls", "companies": "datasets/indname.xls", "roe_us": "datasets/roe.xls", "roe_global": "datasets/roeGlobal.xls", "country": "datasets/ctrypremJuly26.xlsx", "erp_monthly": "implprem/ERPbymonth.xlsx", } # US listings in his company file (exchange prefixes); everything else is a non-US listing US_EXCHANGES = ("NYSE", "NasdaqGS", "NasdaqGM", "NasdaqCM", "NYSEAM", "ARCA", "BATS", "OTCPK", "OTCQX", "OTCQB", "NYSEArca") _SUFFIX = re.compile(r"\b(inc|corp|corporation|co|company|ltd|limited|plc|sa|ag|nv|se|holdings?|group|the|llc|lp|" r"incorporated|n\.v|s\.a|a/s|asa|spa|s\.p\.a|kgaa|oyj|ab|bhd|tbk|pcl)\b\.?", re.I) def _norm_name(name): n = re.sub(r"\([^)]*\)", " ", str(name or "")) n = _SUFFIX.sub(" ", n.replace("&", " and ")) return re.sub(r"[^a-z0-9]+", " ", n.lower()).strip() # ═══════════════════════════════════════════════════════════════════════════════════════ # builder (local / refresh chain) # ═══════════════════════════════════════════════════════════════════════════════════════ def _num(v): try: if v is None: return None if isinstance(v, str): pct = "%" in v v = v.strip().replace("%", "").replace(",", ".") x = float(v) / 100.0 if pct else float(v) else: x = float(v) return None if x != x else x except (TypeError, ValueError): return None def _industry_table(path): import pandas as pd raw = pd.read_excel(path, sheet_name="Industry Averages", header=None) h = raw.index[raw.iloc[:, 0].astype(str).str.strip().eq("Industry Name")][0] df = raw.iloc[h + 1:].copy() df.columns = [str(c).strip() for c in raw.iloc[h]] date = raw.iloc[0, 1] return df.dropna(subset=["Industry Name"]), (str(date)[:10] if date is not None else None) def build(src_dir, out=DATA_FILE): import pandas as pd p = lambda k: os.path.join(src_dir, os.path.basename(SOURCES[k])) data = {"meta": {"source": "Aswath Damodaran, https://pages.stern.nyu.edu/~adamodar/", "files": {k: BASE_URL + v for k, v in SOURCES.items()}, "built_at": _dt.date.today().isoformat()}} # implied ERP by month (his headline figure: trailing-12-month cash yield, column "ERP (T12m)") e = pd.read_excel(p("erp_monthly"), sheet_name="Historical ERP") rows = [] for _, r in e.iterrows(): d, v = r.get("Start of month"), _num(r.get("ERP (T12m)")) if d is not None and v is not None and str(d) != "NaT": rows.append({"month": str(d)[:10], "erp": round(v, 5), "tbond": _num(r.get("T.Bond Rate"))}) data["erp"] = {"measure": "implied ERP, trailing 12-month cash yield (ERP (T12m))", "monthly": rows[-36:]} # country risk premiums (rating-based default spread × relative equity volatility) c = pd.read_excel(p("country"), sheet_name="ERPs by country", header=None) h = c.index[c.iloc[:, 0].astype(str).str.strip().eq("Country")][0] mature = _num(c.iloc[2, 4]) countries = {} for _, r in c.iloc[h + 1:].iterrows(): name = str(r.iloc[0]).strip() if not name or name == "nan" or _num(r.iloc[5]) is None: continue countries[name] = {"region": str(r.iloc[1]), "rating": str(r.iloc[2]), "default_spread": _num(r.iloc[3]), "total_erp": _num(r.iloc[4]), "crp": _num(r.iloc[5])} data["country"] = {"as_of": str(c.iloc[1, 1])[:10], "mature_erp": mature, "countries": countries} # industries: US and global tables inds = {} market = {} for region, keys in (("US", ("betas_us", "capex_us", "margin_us", "ps_us", "wacc_us", "pe_us", "roe_us")), ("Global", ("betas_global", "capex_global", "margin_global", "ps_global", "wacc_global", "pe_global", "roe_global"))): b, bd = _industry_table(p(keys[0])); cx, _ = _industry_table(p(keys[1])); m, _ = _industry_table(p(keys[2])) ps, _ = _industry_table(p(keys[3])); wc, wd = _industry_table(p(keys[4])); pe, _ = _industry_table(p(keys[5])) roe, _ = _industry_table(p(keys[6])) cx = cx.set_index("Industry Name"); m = m.set_index("Industry Name"); ps = ps.set_index("Industry Name") pe = pe.set_index("Industry Name"); roe = roe.set_index("Industry Name") # the mature company: his "Total Market (without financials)" row of the cost-of-capital file, with # the Treasury rate and marginal tax rate he estimated it at (so it can be rebuilt at today's rates) wraw = pd.read_excel(p(keys[4]), sheet_name="Industry Averages", header=None) def lab(s): # the first number to the right of a label on the input panel for _, r in wraw.iterrows(): if str(r.iloc[0]).strip().startswith(s): return next((_num(v) for v in list(r.iloc[1:8]) if _num(v) is not None), None) return None tm = wc.set_index("Industry Name").loc["Total Market (without financials)"] rf_then = lab("Long Term Treasury bond rate") market[region] = {"as_of": wd, "row": "Total Market (without financials)", "beta": _num(tm.get("Beta")), "debt_weight": _num(tm.get("D/(D+E)")), "pre_tax_cost_of_debt": _num(tm.get("Cost of Debt")), "default_spread": (_num(tm.get("Cost of Debt")) - rf_then) if rf_then is not None else None, "marginal_tax": lab("If yes, enter the marginal tax rate"), "rf": rf_then, "erp": lab("Risk Premium to Use for Equity"), "cost_of_capital": _num(tm.get("Cost of Capital")), "n_firms": _num(tm.get("Number of Firms"))} tab = {} for _, r in b.iterrows(): n = str(r["Industry Name"]).strip() row = {"n_firms": _num(r.get("Number of firms")), "levered_beta": _num(r.get("Beta")), "d_e": _num(r.get("D/E Ratio")), "tax_rate": _num(r.get("Effective Tax rate")), "unlevered_beta": _num(r.get("Unlevered beta")), "unlevered_beta_cash": _num(r.get("Unlevered beta corrected for cash"))} if n in cx.index: row["sales_to_capital"] = _num(cx.loc[n].get("Sales/ Invested Capital (LTM)")) if n in m.index: row["operating_margin"] = _num(m.loc[n].get("Pre-tax Unadjusted Operating Margin")) row["operating_margin_adj"] = _num(m.loc[n].get("Pre-tax Lease & R&D adj Margin")) row["net_margin"] = _num(m.loc[n].get("Net Margin")) if n in ps.index: row["ev_sales"] = _num(ps.loc[n].get("EV/Sales")) row["price_sales"] = _num(ps.loc[n].get("Price/Sales")) if n in pe.index: # aggregate market cap ÷ trailing net income of the profitable firms: robust to loss-makers, unlike # the average of firm P/Es (Software averages 122x with 70% of firms losing money) row["pe_aggregate_profitable"] = _num(pe.loc[n].get("Aggregate Mkt Cap/ Trailing Net Income (only money making firms)")) row["pe_forward"] = _num(pe.loc[n].get("Forward PE")) row["expected_growth_5y"] = _num(pe.loc[n].get("Expected growth - next 5 years")) if n in roe.index: row["roe"] = _num(roe.loc[n].get("ROE (unadjusted)")) tab[n] = row inds[region] = {"as_of": bd, "industries": tab} data["industries"] = inds data["market"] = market # companies → industry and country (his classification) comp = pd.read_excel(p("companies"), sheet_name="By company name") companies, names, sic_counts = {}, {}, {} for _, r in comp.iterrows(): key, ind = str(r.get("Exchange:Ticker") or "").strip(), str(r.get("Industry Group") or "").strip() if not key or not ind or ind == "nan": continue country = str(r.get("Country") or "").strip() sic = str(r.get("SIC Code") or "").split(".")[0] companies[key] = [ind, country] nm = _norm_name(r.get("Company Name")) if nm and nm not in names: names[nm] = key if country == "United States" and sic.isdigit(): sic_counts.setdefault(sic, {}).setdefault(ind, 0) sic_counts[sic][ind] += 1 data["companies"] = companies data["names"] = names data["sic_industry"] = {s: max(v.items(), key=lambda kv: kv[1])[0] for s, v in sic_counts.items()} with gzip.open(out, "wt", encoding="utf-8") as fh: json.dump(data, fh, separators=(",", ":")) return {"erp_months": len(rows), "countries": len(countries), "companies": len(companies), "industries_us": len(inds["US"]["industries"]), "industries_global": len(inds["Global"]["industries"])} # ═══════════════════════════════════════════════════════════════════════════════════════ # loader (engine) # ═══════════════════════════════════════════════════════════════════════════════════════ _CACHE = {"data": None, "loaded": False} OVERRIDE_FILE = os.path.join(os.getenv("CRUCIBLE_CACHE") or os.path.dirname(os.path.abspath(__file__)), "damodaran_erp_monthly.json") def _erp_rows(): rows = list((_data() or {}).get("erp", {}).get("monthly") or []) try: with open(OVERRIDE_FILE) as fh: extra = json.load(fh).get("monthly") or [] by = {r["month"]: r for r in rows} by.update({r["month"]: r for r in extra}) rows = [by[k] for k in sorted(by)] except (OSError, ValueError): pass return rows def refresh_erp_if_stale(max_age_days=35, timeout=30): """His implied ERP is published monthly. When the newest month we hold is older than ``max_age_days``, fetch his monthly file and cache the parsed rows (the bundle stays as is). Returns a short status string; a failed fetch leaves the bundled months in use.""" rows = _erp_rows() if rows: age = (_dt.date.today() - _dt.date.fromisoformat(rows[-1]["month"][:10])).days if age <= max_age_days: return f"current ({rows[-1]['month']})" try: import io import pandas as pd import requests r = requests.get(BASE_URL + SOURCES["erp_monthly"], timeout=timeout) r.raise_for_status() e = pd.read_excel(io.BytesIO(r.content), sheet_name="Historical ERP") new = [] for _, x in e.iterrows(): d, v = x.get("Start of month"), _num(x.get("ERP (T12m)")) if d is not None and v is not None and str(d) != "NaT": new.append({"month": str(d)[:10], "erp": round(v, 5), "tbond": _num(x.get("T.Bond Rate"))}) with open(OVERRIDE_FILE, "w") as fh: json.dump({"fetched": _dt.date.today().isoformat(), "monthly": new[-36:]}, fh) return f"refreshed to {new[-1]['month']}" if new else "fetched, no rows" except Exception as ex: # noqa: BLE001 return f"refresh failed ({str(ex)[:80]}); using {rows[-1]['month'] if rows else 'none'}" def _data(): if not _CACHE["loaded"]: _CACHE["loaded"] = True try: with gzip.open(DATA_FILE, "rt", encoding="utf-8") as fh: _CACHE["data"] = json.load(fh) except (OSError, ValueError): _CACHE["data"] = None return _CACHE["data"] def available(): return _data() is not None def implied_erp(on=None): """{value, month, measure} — his implied ERP for the latest month at or before ``on``.""" d = _data() if not d: return None rows = _erp_rows() if on: rows = [r for r in rows if r["month"] <= str(on)[:10]] or rows r = rows[-1] return {"value": r["erp"], "month": r["month"], "measure": d["erp"]["measure"], "source": d["meta"]["files"]["erp_monthly"]} _COUNTRY_ALIASES = {"USA": "United States", "US": "United States", "UK": "United Kingdom", "Korea": "Korea", "South Korea": "Korea", "Republic of Korea": "Korea", "Hong Kong SAR": "Hong Kong", "Netherlands (the)": "Netherlands"} def country_premium(country): """{crp, rating, total_erp, as_of} for a country of operations, or None when not in his table.""" d = _data() if not d or not country: return None name = _COUNTRY_ALIASES.get(str(country).strip(), str(country).strip()) row = d["country"]["countries"].get(name) if row is None: low = name.lower() row = next((v for k, v in d["country"]["countries"].items() if k.lower() == low or k.lower().startswith(low)), None) return dict(row, country=name, as_of=d["country"]["as_of"], mature_erp=d["country"]["mature_erp"]) if row else None def company_industry(ticker, name=None, sic=None): """(industry, country, how) from his company list: the US listing ticker first, then the company name (foreign issuers are listed under their home ticker), then the SIC code's modal industry.""" d = _data() if not d: return None, None, None t = str(ticker or "").upper() for ex in US_EXCHANGES: hit = d["companies"].get(f"{ex}:{t}") if hit: return hit[0], hit[1], f"ticker {ex}:{t}" nm = _norm_name(name) if nm and nm in d["names"]: k = d["names"][nm] ind, ctry = d["companies"][k] return ind, ctry, f"company name → {k}" s = str(sic or "").split(".")[0] if s and s in d["sic_industry"]: return d["sic_industry"][s], None, f"SIC {s} (modal industry)" return None, None, None def industry_stats(industry, region="US"): d = _data() if not d or not industry: return None tab = d["industries"].get(region) or d["industries"]["US"] row = tab["industries"].get(industry) return dict(row, industry=industry, region=region, as_of=tab["as_of"]) if row else None def bottom_up_beta(stats, debt, market_cap, tax=0.25): """His bottom-up beta: the industry's unlevered beta corrected for cash, re-levered at the company's market debt-to-equity: βL = βu × (1 + (1 − t) × D/E).""" if not stats: return None bu = stats.get("unlevered_beta_cash") or stats.get("unlevered_beta") if bu is None: return None de = (float(debt) / float(market_cap)) if (debt and market_cap and market_cap > 0) else 0.0 return float(bu) * (1.0 + (1.0 - tax) * min(max(de, 0.0), 5.0)) def company_erp(country, on=None): """The ERP for a company operating in ``country``: his implied (mature-market) ERP plus the country risk premium of the country of operations (zero for the US and Aaa countries).""" base = implied_erp(on) if base is None: return None cp = country_premium(country) if country else None us = str(country or "United States").strip() in ("United States", "USA", "US", "") crp = 0.0 if (us or cp is None) else float(cp.get("crp") or 0.0) return {"erp": base["value"] + crp, "implied_erp": base["value"], "crp": crp, "country": (cp or {}).get("country") or country, "rating": (cp or {}).get("rating"), "erp_month": base["month"], "crp_as_of": (cp or {}).get("as_of"), "basis": ("implied ERP + country risk premium of the country of operations" if crp else "implied ERP (mature market)")} def industry_multiples(industry, region="US"): """Sourced industry pricing anchors: {pe: {value, basis, source, date, universe}, ps: {...}} or {}.""" st = industry_stats(industry, region) or industry_stats(industry, "US") or {} d = _data() or {} out = {} if st.get("pe_aggregate_profitable"): out["pe"] = {"value": st["pe_aggregate_profitable"], "basis": "TRAILING_GAAP (aggregate market cap ÷ trailing net income, " "profitable firms)", "source": "Damodaran pedata/peGlobal", "date": st.get("as_of"), "universe": f"{st.get('region')} {industry} ({int(st.get('n_firms') or 0)} firms)"} if st.get("price_sales"): out["ps"] = {"value": st["price_sales"], "basis": "TRAILING (aggregate market cap ÷ trailing revenue)", "source": "Damodaran psdata/psGlobal", "date": st.get("as_of"), "universe": f"{st.get('region')} {industry} ({int(st.get('n_firms') or 0)} firms)"} return out # industries too broad or economically mixed for their beta to stand as company-specific terminal evidence BROAD_INDUSTRIES = ("Diversified", "Business & Consumer Services", "Conglomerates", "Retail (General)", "Financial Svcs. (Non-bank & Insurance)", "Information Services", "Total Market") _FINANCIAL_WORDS = ("bank", "financial svcs", "insurance", "brokerage", "investments & asset", "reit") BETA_SANE = (0.5, 2.0) def beta_quality(inputs, is_financial=False): """{status: HIGH|LOW, reasons[], beta, beta_source, beta_industry}: may this beta stand as evidence of the company's own risk in stable growth? Checks the industry mapping, the business-model fit of the industry, the business mix and the beta's statistical sanity. Peer betas are not in the data, so peer consistency is recorded as not assessed.""" di = inputs or {} ind = str(di.get("industry") or "") b = di.get("beta") reasons = [] if not ind: reasons.append("no industry classification") if di.get("industry_match") and not str(di.get("industry_match")).startswith("ticker"): reasons.append(f"industry matched by {di.get('industry_match')}, not by ticker") if ind in BROAD_INDUSTRIES: reasons.append(f"industry '{ind}' is broad or economically mixed") if not is_financial and any(w in ind.lower() for w in _FINANCIAL_WORDS): reasons.append(f"a financial-services industry beta ('{ind}') for a non-financial business: " "financial firms' debt is operating, so their unlevered betas are not comparable") if b is None or not (BETA_SANE[0] <= float(b) <= BETA_SANE[1]): reasons.append(f"beta {b if b is None else round(float(b), 2)} outside the plausible range {BETA_SANE[0]}–{BETA_SANE[1]}") mix = di.get("business_mix") or [] if len({r.get("industry") for r in mix}) > 2: reasons.append("business mix spans more than two industries") # every industry the business-mix beta used must come through a compatible mapping (not only the company's own) mix_bad = business_mix_problems(mix, ind or None, is_financial) reasons += [f"business-mix segment mapping incompatible: {x}" for x in mix_bad] return {"status": "LOW" if reasons else "HIGH", "reasons": reasons, "beta": b, "business_mix_beta_valid": (not mix_bad) if mix else None, "beta_source": di.get("beta_basis"), "beta_industry": ind or None, "peer_consistency": "not assessed (no peer betas in the data)"} def mature_cost_of_capital(rf, erp, region="US", tax=None): """The cost of capital of a mature company, his stable-growth convention ("roughly that of an average company"): the beta, debt ratio and default spread of his Total Market (without financials) row, rebuilt at today's riskfree rate and the company's ERP. ``None`` when the dataset lacks the row.""" d = _data() mk = ((d or {}).get("market") or {}).get(region) or ((d or {}).get("market") or {}).get("US") if not mk or rf is None or erp is None or mk.get("beta") is None or mk.get("debt_weight") is None: return None t = mk.get("marginal_tax") if tax is None else tax t = 0.25 if t is None else float(t) ke = float(rf) + float(mk["beta"]) * float(erp) kd = (float(rf) + float(mk.get("default_spread") or 0.0)) * (1.0 - t) w = float(mk["debt_weight"]) return {"value": (1.0 - w) * ke + w * kd, "cost_of_equity": ke, "after_tax_cost_of_debt": kd, "beta": mk["beta"], "debt_weight": w, "default_spread": mk.get("default_spread"), "tax": t, "region": region, "as_of": mk.get("as_of"), "basis": f"mature company: {mk['row']} ({region}, {mk.get('as_of')}) at today's riskfree rate and ERP"} def download(dest): """Fetch every source file into ``dest`` (the refresh chain calls this, then ``build``).""" import requests os.makedirs(dest, exist_ok=True) got = {} for k, path in SOURCES.items(): r = requests.get(BASE_URL + path, timeout=120) r.raise_for_status() with open(os.path.join(dest, os.path.basename(path)), "wb") as fh: fh.write(r.content) got[k] = {"bytes": len(r.content), "last_modified": r.headers.get("last-modified")} return got if __name__ == "__main__": import sys dest = sys.argv[1] if len(sys.argv) > 1 else "damodaran_src" print(json.dumps(download(dest), indent=1)) print(json.dumps(build(dest), indent=1)) # ── segments → his industries (deterministic; first compatible match wins; unmatched → the company's industry) ── def _kw(*tokens, stems=()): """One keyword pattern. Every TOKEN matches only as a whole word (bounded on both sides: "card" never matches "cardiovascular", "gas" never "gastroenterology", "power" never "powertrain"). A STEM is declared explicitly and matches a word that begins with it ("pharma" → pharmaceutical).""" parts = list(tokens) + [s + r"\w*" for s in stems] return r"\b(?:" + "|".join(parts) + r")\b" SEGMENT_INDUSTRY = [ (_kw("launch(?:es)?", "space", "rockets?", "aerospace", "defen[cs]e", "aircraft", "aviation", "avionics"), "Aerospace/Defense"), (_kw("connectivity", "broadband", "starlink", "wireless", "internet services?", "satellite internet", stems=("telecom",)), "Telecom. Services"), (_kw("cloud", "aws", "azure", "software", "saas", "subscription software", "productivity", "business processes"), "Software (System & Application)"), (_kw("ai", "artificial intelligence", "machine learning", "llm", "generative"), "Software (System & Application)"), (_kw(stems=("advertis",)), "Advertising"), (_kw("search", "social", "family of apps", "online services"), "Software (Internet)"), # "retail" alone is not a business (retail electricity, retail banking): only store/merchandise retail maps here (_kw("online stores?", "physical stores?", "retail stores?", "e-?commerce", "third[- ]party sellers?", "merchandise", "general merchandise"), "Retail (General)"), (_kw("semiconductors?", "chips?", "gpus?", "compute and networking", "data cent(?:er|re)s?"), "Semiconductor"), (_kw("automotive", "vehicles?", "auto"), "Auto & Truck"), (_kw("energy generation", "energy storage", "solar", "renewables?"), "Green & Renewable Energy"), (_kw("drugs?", "oncology", "vaccines?", stems=("pharma", "biotech", "therapeut")), "Drugs (Pharmaceutical)"), (_kw("medical", "medtech", "diagnostics?", "devices? health", "healthcare products"), "Healthcare Products"), (_kw("gaming", "video games?", "xbox"), "Software (Entertainment)"), (_kw("media", "entertainment", "studios?", "streaming", "content", "parks"), "Entertainment"), (_kw("cable"), "Cable TV"), (_kw("beverages?"), "Beverage (Soft)"), (_kw("foods?", "snacks?"), "Food Processing"), (_kw("hotels?", "casinos?", "resorts?", "cruises?"), "Hotel/Gaming"), (_kw("restaurants?"), "Restaurant/Dining"), (_kw("oil", "gas", "upstream", "downstream", "refining", "chemicals? and energy"), "Oil/Gas (Integrated)"), (_kw("chemicals?"), "Chemical (Diversified)"), (_kw("mining", "metals", "copper", "gold"), "Metals & Mining"), (_kw("machinery", "industrial automation", "automation", "motion"), "Machinery"), (_kw("buildings?", "hvac", "construction"), "Building Materials"), (_kw("power", "electric(?:ity)?", "utility", "utilities", "retail energy", "wholesale energy", "energy charge"), "Power"), (_kw("iphone", "mac", "ipad", "wearables", "computers?", "pcs?", "hardware", "personal computing"), "Computers/Peripherals"), (_kw("insurance"), "Insurance (General)"), (_kw("payments?", "cards?"), "Financial Svcs. (Non-bank & Insurance)"), ] _IGNORE_SEGMENTS = re.compile(r"\b(corporate|eliminations?|unallocated|reconcil)", re.I) # a segment named only by a place is a region of the company's own business, not a different business _REGION_NAME = re.compile(r"^(?:(?:north|south|east|west|central|northeast|northwest|southeast|southwest|midwest|mid-?atlantic|gulf|" r"pacific|atlantic|americas?|latin|europe(?:an)?|emea|asia|apac|greater|china|japan|india|canada|mexico|" r"texas|california|florida|new york|international|domestic|united|states|kingdom|u\.?s\.?|uk|rest|of|the|" r"world|and|other|region|regions|markets?|operations|segment|&|\s|-)+)$", re.I) def keyword_matches(text): """Every (industry, pattern, matched words) the text matches, in table order.""" low = str(text or "").lower() out = [] for rx, ind in SEGMENT_INDUSTRY: m = re.search(rx, low) if m: out.append((ind, rx, m.group(0))) return out def segment_industry(name, product_lines=(), fallback=None): """(industry, how) for one reportable segment: its name first, then its product-line names. A segment named only by a region is the company's own business in that region.""" if _REGION_NAME.match(str(name or "").strip()): return fallback, "regional segment: company industry" for text, how in ((name, "segment name"), (" ".join(product_lines or ()), "product lines")): hits = keyword_matches(text) if hits: return hits[0][0], f"{how} matches {hits[0][0]}" return fallback, "no keyword match: company industry" # ── parent-sector compatibility: what a segment's industry may be, given the company it belongs to ── PASS, REJECTED, UNAVAILABLE_MAPPING = "PASS", "REJECTED", "UNAVAILABLE" # explicit evidence that a segment is itself a financial business: one of these stands as a whole word in its own text _FINANCIAL_EVIDENCE = _kw("financial", "finance", "financing", "credit", "bank", "banking", "lending", "leasing", "insurance", "reinsurance", "brokerage", "payments?", "cards?", "asset management", "wealth management") def is_financial_industry(industry): return any(w in str(industry or "").lower() for w in _FINANCIAL_WORDS) def industry_compatible(parent_industry, parent_is_financial, industry, segment_text): """(ok, reason): may a segment of this company carry this industry? A clearly non-financial parent's segment may carry a financial industry only on explicit evidence in the segment's own text (a financial word standing as a whole word): financial firms' debt is operating, so their EV/Sales and unlevered betas do not describe an operating business. No other parent/segment pair is restricted in this pass.""" if not industry: return False, "no industry" parent_fin = bool(parent_is_financial) or is_financial_industry(parent_industry) if is_financial_industry(industry) and not parent_fin: if not re.search(_FINANCIAL_EVIDENCE, str(segment_text or "").lower()): return False, (f"financial industry '{industry}' for a segment of a non-financial company " f"({parent_industry or 'unclassified'}) with no financial business named in its own text") return True, "" def map_segment(segment_id, name, product_lines=(), parent_industry=None, parent_is_financial=False, use_product_lines=True): """SegmentIndustryMapping for one reportable segment. Fallback hierarchy after a rejection: 1. the next keyword match in the segment's own text that is compatible with the parent (a validated segment industry); 2. the company-level industry (compatible with itself); 3. a broader compatible sector: not available (his datasets carry no sector level above the industry); 4. mapping unavailable (no selected industry: the segment may feed neither value weights nor betas).""" text = str(name or "") m = {"segment_id": segment_id, "input_text": text, "selected_industry": None, "mapping_method": None, "matched_pattern": None, "matched_text": None, "parent_industry": parent_industry, "compatibility_status": UNAVAILABLE_MAPPING, "fallback_used": None, "rejected": [], "eligible_for_value_weight": False, "eligible_for_beta": False} def select(industry, method, pattern=None, matched=None, fallback=None): m.update(selected_industry=industry, mapping_method=method, matched_pattern=pattern, matched_text=matched, compatibility_status=PASS, fallback_used=fallback, eligible_for_value_weight=True, eligible_for_beta=True) return m if _REGION_NAME.match(text.strip()): if parent_industry: return select(parent_industry, "regional segment: company industry", fallback="COMPANY_INDUSTRY") m["mapping_method"] = "regional segment: no company industry" return m sources = [(text, "segment name")] if use_product_lines and product_lines: sources.append((" ".join(str(x) for x in product_lines if x), "product lines")) for src, how in sources: for ind, rx, matched in keyword_matches(src): ok, why = industry_compatible(parent_industry, parent_is_financial, ind, src) if ok: return select(ind, f"{how} matches {ind}", rx, matched) m["rejected"].append({"industry": ind, "source": how, "matched_pattern": rx, "matched_text": matched, "compatibility_status": REJECTED, "reason": why}) if parent_industry: return select(parent_industry, ("company industry (its keyword match was rejected as incompatible)" if m["rejected"] else "no keyword match: company industry"), fallback="COMPANY_INDUSTRY") m["mapping_method"] = "no compatible industry and no company industry" m["fallback_used"] = "UNAVAILABLE" return m class IncompatibleMapping(Exception): """A segment's industry row was requested through a mapping that did not pass compatibility.""" def mapping_problem(mapping, parent_is_financial=False): """Why a SegmentIndustryMapping may not supply an industry row (None = usable). Its own status must be PASS and the pair it records is re-checked, so a record that merely says PASS cannot carry an incompatible industry.""" mp = mapping or {} if mp.get("compatibility_status") != PASS or not mp.get("selected_industry"): return f"mapping status {mp.get('compatibility_status')}" ok, why = industry_compatible(mp.get("parent_industry"), parent_is_financial, mp.get("selected_industry"), mp.get("input_text")) return None if ok else why def industry_row(mapping, region, use, parent_is_financial=False): """The industry statistics a mapping may supply for ``use`` ("value_weight": EV/Sales; "beta": unlevered beta). The ONE accessor the business mix reads, so a bad mapping cannot be amplified through weight and beta together.""" why = mapping_problem(mapping, parent_is_financial) flag = {"value_weight": "eligible_for_value_weight", "beta": "eligible_for_beta"}[use] if why is None and not (mapping or {}).get(flag): why = f"mapping not eligible for {use}" if why: raise IncompatibleMapping(f"{(mapping or {}).get('segment_id') or (mapping or {}).get('input_text')}: {why}") ind = mapping["selected_industry"] return industry_stats(ind, region) or industry_stats(ind, "US") or {} def business_mix(segments, region, company_industry, debt, market_cap, tax=0.25, parent_is_financial=False): """His business-mix beta: each business line's industry unlevered beta (cash-corrected), weighted by the ESTIMATED VALUE of the business (its revenue × its industry's EV/Sales, his weighting), and re-levered at the company's market D/E. Revenue weights are used only when an EV/Sales is missing, and the basis says so. Also returns the industry sales-to-capital of the mix (capital = Σ revenue / sales-to-capital) and each segment's industry benchmarks. None unless 2+ business lines. Every row reads its industry through a SegmentIndustryMapping that passed compatibility (industry_row): the same industry supplies the EV/Sales weight AND the beta, so an incompatible mapping is refused for both and the whole mix is withheld (the single-industry beta then applies).""" rows = [] for g in segments or []: if _IGNORE_SEGMENTS.search(str(g.get("name") or "")) or not (g.get("revenue") or 0) > 0: continue if g.get("allocations") is not None: # canonical segments (segments.allocate): one row per industry allocation, fractions sum to 1 allocs = [(a["industry"], a["basis"], float(a["fraction"]), a.get("mapping") or {"segment_id": g.get("segment_id"), "input_text": g.get("name"), "selected_industry": a["industry"], "parent_industry": company_industry, "compatibility_status": PASS, "eligible_for_value_weight": True, "eligible_for_beta": True}) for a in g["allocations"]] else: if g.get("kind") == "geographic": continue mp = map_segment(g.get("segment_id"), g.get("name"), [p.get("name") for p in g.get("product_lines") or []], company_industry, parent_is_financial) if not mp["selected_industry"]: return None allocs = [(mp["selected_industry"], mp["mapping_method"], 1.0, mp)] for ind, how, fr, mp in allocs: try: st_w = industry_row(mp, region, "value_weight", parent_is_financial) st_b = industry_row(mp, region, "beta", parent_is_financial) except IncompatibleMapping: return None # never a precision blend over a mapping that failed compatibility rows.append({"name": g.get("name") + (f" ({ind})" if len(allocs) > 1 else ""), "segment_id": g.get("segment_id"), "fraction": fr, "revenue": g["revenue"] * fr, "industry": ind, "match": how, "unlevered_beta": st_b.get("unlevered_beta_cash") or st_b.get("unlevered_beta"), "industry_sales_to_capital": st_b.get("sales_to_capital"), "industry_operating_margin": st_b.get("operating_margin"), "industry_ev_sales": st_w.get("ev_sales"), "mapping": {k: mp.get(k) for k in ("segment_id", "input_text", "selected_industry", "mapping_method", "matched_pattern", "matched_text", "parent_industry", "compatibility_status", "fallback_used", "eligible_for_value_weight", "eligible_for_beta")}}) if len(rows) < 2 or any(r["unlevered_beta"] is None for r in rows) or len({r["industry"] for r in rows}) < 2: return None # one business, or several lines of the same business: the single-industry beta applies tot = sum(r["revenue"] for r in rows) by_value = all((r["industry_ev_sales"] or 0) > 0 for r in rows) for r in rows: r["revenue_weight"] = r["revenue"] / tot r["estimated_value"] = (r["revenue"] * r["industry_ev_sales"]) if by_value else None vtot = sum(r["estimated_value"] for r in rows) if by_value else None for r in rows: r["weight"] = (r["estimated_value"] / vtot) if by_value else r["revenue_weight"] bu = sum(r["weight"] * r["unlevered_beta"] for r in rows) s2c = [r for r in rows if r["industry_sales_to_capital"]] s2c_mix = (sum(r["revenue"] for r in s2c) / sum(r["revenue"] / r["industry_sales_to_capital"] for r in s2c)) if s2c else None de = (float(debt) / float(market_cap)) if (debt and market_cap and market_cap > 0) else 0.0 return {"segments": rows, "unlevered_beta": bu, "beta": bu * (1.0 + (1.0 - tax) * min(max(de, 0.0), 5.0)), "sales_to_capital": s2c_mix, "weighting": "value" if by_value else "revenue", "weights": ("estimated value: segment revenue × its industry EV/Sales (his datasets)" if by_value else "share of revenue (an industry EV/Sales is missing, so value weights could not be formed)")} def business_mix_problems(mix_rows, company_industry, parent_is_financial=False): """[problem] for the rows of a business-mix beta: every industry it used must come through a compatible mapping (BUSINESS_MIX_BETA_VALID). A row with no mapping record is re-checked from its own industry and name.""" bad = [] for r in mix_rows or []: mp = r.get("mapping") or {"segment_id": r.get("segment_id"), "input_text": str(r.get("name") or "").split(" (")[0], "selected_industry": r.get("industry"), "parent_industry": company_industry, "compatibility_status": PASS} mp = dict(mp, selected_industry=r.get("industry")) # the industry the row actually used why = mapping_problem(mp, parent_is_financial) if why: bad.append(f"{r.get('name')}: {why}") return bad # Different economics (his reason for valuing a company business by business): reported operating # margins far apart, or businesses whose industries carry very different risk. SEGMENT_MARGIN_SPREAD = 0.20 SEGMENT_BETA_SPREAD = 0.25 def segment_build_evidence(filed_segments, mix_rows=None): """{required, kind, reason, low, high, low_label, high_label}. A segment-by-segment story is required when the company files two or more business lines whose reported operating margins span at least SEGMENT_MARGIN_SPREAD ("margin"), or whose industries' unlevered betas span at least SEGMENT_BETA_SPREAD ("risk"). The numbers are kept as fields so a report can cite them as claims.""" segs = [g for g in filed_segments or [] if g.get("kind", "business_line") == "business_line"] if len(segs) < 2: return {"required": False, "kind": None, "reason": "fewer than two business lines filed"} m = [(g.get("name"), g["margin"]) for g in segs if g.get("margin") is not None] if len(m) >= 2: lo, hi = min(m, key=lambda x: x[1]), max(m, key=lambda x: x[1]) if hi[1] - lo[1] >= SEGMENT_MARGIN_SPREAD: return {"required": True, "kind": "margin", "low": lo[1], "high": hi[1], "low_label": lo[0], "high_label": hi[0], "reason": f"reported operating margins span {lo[1]*100:.0f}% ({lo[0]}) to {hi[1]*100:.0f}% ({hi[0]})"} # risk is compared between REPORTABLE SEGMENTS: a segment allocated to several industries (segments.allocate) # carries its fraction-weighted beta, so a product-line split inside one segment is not a second business per = {} for r in mix_rows or []: if r.get("unlevered_beta") is None: continue k = r.get("segment_id") or r.get("name") or id(r) a = per.setdefault(k, {"label": r.get("industry"), "w": 0.0, "bw": 0.0, "n": 0}) fr = float(r.get("fraction") or 1.0) a["w"] += fr; a["bw"] += fr * r["unlevered_beta"]; a["n"] += 1 if a["n"] > 1: a["label"] = str(r.get("name") or k).split(" (")[0] + " (mixed)" b = [(a["label"], a["bw"] / a["w"]) for a in per.values() if a["w"] > 0] if len(b) >= 2: lo, hi = min(b, key=lambda x: x[1]), max(b, key=lambda x: x[1]) if hi[1] - lo[1] >= SEGMENT_BETA_SPREAD: return {"required": True, "kind": "risk", "low": lo[1], "high": hi[1], "low_label": lo[0], "high_label": hi[0], "reason": f"industry risk differs: {lo[0]} unlevered beta {lo[1]:.2f} vs {hi[0]} {hi[1]:.2f}"} return {"required": False, "kind": None, "reason": "business lines have similar margins and industry risk"} def segment_build_required(filed_segments, mix_rows=None): """(required, reason) — see ``segment_build_evidence``.""" e = segment_build_evidence(filed_segments, mix_rows) return e["required"], e["reason"]