Damodaran L1.5 데이터 감사

공개 호출 방식

import dartlab
import importlib.resources as resources
import json
from pathlib import Path

import polars as pl
from dartlab.synth.damodaranL15 import buildDamodaranMemo

target = "005930"
c = dartlab.Company(target)
market = getattr(c, "market", "US" if not target.isdigit() else "KR")
currency = getattr(c, "currency", "USD" if market == "US" else "KRW")
company_name = getattr(c, "corpName", getattr(c, "companyName", target))


def _loadReference(name):
    return json.loads(resources.files("dartlab.reference.data").joinpath(name).read_text(encoding="utf-8"))


def _safeShow(topic):
    try:
        table = c.show(topic, freq="Y")
    except TypeError:
        table = c.show(topic)
    except Exception:
        return pl.DataFrame()
    return table if isinstance(table, pl.DataFrame) else pl.DataFrame()


def _latestPrice(frame):
    if not isinstance(frame, pl.DataFrame) or frame.height == 0:
        return {}
    date_col = "date" if "date" in frame.columns else "Date" if "Date" in frame.columns else None
    close_col = "close" if "close" in frame.columns else "Close" if "Close" in frame.columns else None
    latest = frame.sort(date_col).tail(1).to_dicts()[0] if date_col else frame.tail(1).to_dicts()[0]
    out = {}
    if close_col and latest.get(close_col) is not None:
        out["price"] = latest.get(close_col)
    if date_col and latest.get(date_col) is not None:
        out["priceDate"] = str(latest.get(date_col))
    return out


def _marketData():
    out = {}
    try:
        price_frame = dartlab.gather("price", target, market="US") if market == "US" else dartlab.gather("price", target)
        out.update(_latestPrice(price_frame))
    except Exception as exc:
        out["priceError"] = type(exc).__name__

    if market == "KR":
        krx_path = Path("data/krx/prices/raw-2026.parquet")
        if krx_path.exists():
            try:
                krx = (
                    pl.scan_parquet(str(krx_path))
                    .filter(pl.col("ISU_CD") == target)
                    .select(["BAS_DD", "TDD_CLSPRC", "MKTCAP", "LIST_SHRS"])
                    .sort("BAS_DD")
                    .tail(1)
                    .collect()
                )
                if krx.height:
                    row = krx.to_dicts()[0]
                    out.update(
                        {
                            "price": row.get("TDD_CLSPRC") or out.get("price"),
                            "priceDate": str(row.get("BAS_DD") or out.get("priceDate")),
                            "marketCap": row.get("MKTCAP"),
                            "shares": row.get("LIST_SHRS"),
                        }
                    )
            except Exception as exc:
                out["marketCapError"] = type(exc).__name__

    if market == "US" and out.get("price") is not None:
        cik = str(getattr(c, "cik", "") or "")
        for path in (Path(f"data/edgar/finance/{cik}.parquet"), Path(f"data/edgar/finance/{target}.parquet")):
            if not path.exists():
                continue
            try:
                shares = (
                    pl.scan_parquet(str(path))
                    .filter((pl.col("unit") == "shares") & pl.col("tag").str.contains("SharesOutstanding"))
                    .select(["val", "filed"])
                    .sort("filed")
                    .tail(1)
                    .collect()
                )
                if shares.height:
                    out["shares"] = shares["val"][0]
                    out["marketCap"] = float(out["price"]) * float(out["shares"])
                    break
            except Exception as exc:
                out["marketCapError"] = type(exc).__name__
    return out


country_defaults = _loadReference("damodaranDefaults.json")
industry_defaults = _loadReference("damodaranIndustryDefaults.json")
statements = {topic: _safeShow(topic) for topic in ("IS", "BS", "CF")}
memo = buildDamodaranMemo(
    target=target,
    market=market,
    currency=currency,
    companyName=company_name,
    statements=statements,
    countryDefaults=country_defaults,
    industryDefaults=industry_defaults,
    marketData=_marketData(),
)

emit_result(
    table=memo["tables"]["dataAudit"],
    values=memo["headline"],
    date=memo.get("asOf"),
    units=memo["units"],
    sources=memo["sources"],
)

호출 동작

1. 결론 도출

대상 기업을 usable, usableWithFallback, blocked 중 하나로 판정한다. usable은 재무 3표, 가격, 국가 reference, 산업 reference가 모두 확인된 경우에만 쓴다.

2. 핵심 근거 수집

Company.show("IS"|"BS"|"CF"|"ratios"|"segments"), dartlab.gather("price"), reference/data/damodaranDefaults.json, reference/data/damodaranIndustryDefaults.json를 확인한다.

3. 메커니즘 분석

데이터 감사는 계산 전 게이트다. 재무 패널이 없으면 정규화가 불가능하고, 가격·시총이 없으면 reverse DCF가 불가능하며, reference가 stale이면 비용자본 가정이 fallback으로 내려간다.

4. 반례·한계

EDGAR는 KR topic alias보다 거칠 수 있다. segments가 없거나 사업 설명 topic이 provider별 이름으로만 있으면 결론을 낮은 confidence로 내려야 한다.

5. 후속 모니터링

country reference as-of, industry reference coverage, price latest date, missing topic list를 다음 단계로 넘긴다.

대표 반환 형태

coverage : list[dict] — area, status, rows, reason을 담는다. 최종 decision은 usable, usableWithFallback, blocked 중 하나다.

연계 절차

1. recipes.fundamental.valuation.damodaran.businessModelFit - 모델 적합성 판정.
2. recipes.fundamental.valuation.damodaran.normalizedFinancials - 재무 패널 정규화.
3. recipes.fundamental.valuation.damodaran.costOfCapital - reference stale/fallback 반영.

기본 검증

• 5개 고정 타깃 중 KR 1개, US 1개 이상은 usableWithFallback 이상이어야 한다.
• 금융업 타깃은 데이터가 있어도 일반 FCFF usable로 승격하지 않는다.
• L2 호출 금지 정적 검사를 통과해야 한다.