| """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_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() |
|
|
|
|
| |
| |
| |
| 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()}} |
| |
| 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:]} |
| |
| 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} |
| |
| 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") |
| |
| |
| wraw = pd.read_excel(p(keys[4]), sheet_name="Industry Averages", header=None) |
| def lab(s): |
| 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: |
| |
| |
| 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 |
| |
| 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"])} |
|
|
|
|
| |
| |
| |
| _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: |
| 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 |
|
|
|
|
| |
| 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") |
| |
| 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)) |
|
|
|
|
| |
| 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)"), |
| |
| (_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) |
| |
| _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" |
|
|
|
|
| |
| PASS, REJECTED, UNAVAILABLE_MAPPING = "PASS", "REJECTED", "UNAVAILABLE" |
| |
| _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: |
| |
| 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 |
| 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 |
| 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")) |
| why = mapping_problem(mp, parent_is_financial) |
| if why: |
| bad.append(f"{r.get('name')}: {why}") |
| return bad |
|
|
|
|
| |
| |
| 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]})"} |
| |
| |
| 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"] |
|
|
