review 1-4 현금흐름 구현 — CF 3구간 + FCF + 이익의 현금 뒷받침

- cashflow.py: calcCashFlowOverview (영업/투자/재무CF + FCF + CF 패턴 시계열)
- cashflow.py: calcCashQuality (영업CF/순이익, 영업CF 마진 시계열)
- cashflow.py: calcCashFlowFlags (영업CF 적자, FCF 적자, 위기형 패턴, 3년 연속 감소)
- builders.py: cashFlowOverviewBlock, cashQualityBlock, cashFlowFlagsBlock
- templates.py: "현금흐름" 템플릿 (partId 1-4) + helper/aiGuide
- registry.py: 현금흐름 블록 등록
- DEV.md: 1-4 현금흐름 구현 완료 반영 (ROIC → 현금흐름으로 변경)
- Part 1 "자본의 여정" 4개 섹션 완성: 수익구조 → 자금조달 → 자산구조 → 현금흐름

Author: eddmpython

src/dartlab/analysis/DEV.md

@@ -133,7 +133,7 @@ gather/는 데이터 수집 엔진이지 분석 엔진이 아니다. 현재 위
 │   ├─ 1-1. 이 회사는 무엇으로 돈을 버는가 (수익 구조)
 │   ├─ 1-2. 돈을 어디서 조달하는가 (자금 구조)
 │   ├─ 1-3. 조달한 돈으로 뭘 준비했는가 (자산 구조)
-│   ├─ 1-4. 투자 대비 얼마를 버는가 (ROIC — 자본의 여정 완결)
+│   ├─ 1-4. 실제로 현금은 어떻게 흘렀는가 (현금흐름)
 │   ├─ 1-5. 경쟁에서 이길 수 있는가 (경쟁 우위)
 │   ├─ 1-6. 산업은 어떤 국면인가 (산업 사이클)
 │   └─ 1-7. 경영진은 신뢰할 수 있는가 (지배구조/ESG)
@@ -185,7 +185,7 @@ gather/는 데이터 수집 엔진이지 분석 엔진이 아니다. 현재 위
 
 | Part | 영역 | 상태 | 비고 |
 |------|------|------|------|
-| 1. 사업 | strategy/ | **1-1, 1-2, 1-3 구현 완료** | 수익구조(9 calc) + 자금구조 + 자산구조 + ESG + Supply |
+| 1. 사업 | strategy/ | **1-1~1-4 구현 완료** | 수익구조 + 자금구조 + 자산구조 + 현금흐름 + ESG + Supply |
 | 2. 검증 | accounting/ | Watch + Signal만 | 이익의 질, Red Flags 없음 |
 | 3. 재무 | financial/ | **완성도 높음** | 10영역 등급 + 부실 예측 + 종합 리포트 |
 | 4. 비교 | comparative/ | 구현됨 | Peer/Sector/Rank/Event |
@@ -200,7 +200,7 @@ gather/는 데이터 수집 엔진이지 분석 엔진이 아니다. 현재 위
 | 1-1 | 수익 구조 | **구현 완료** | strategy/revenue.py |
 | 1-2 | 자금 구조 | **구현 완료** | strategy/capital.py |
 | 1-3 | 자산 구조 | **구현 완료** | strategy/asset.py |
-| 1-4 | ROIC | 설계 완료, 미구현 | — |
+| 1-4 | 현금흐름 | **구현 완료** | strategy/cashflow.py |
 | 3-1 | 이익 구조 | 빌더 있음, 미등록 | strategy/_income_builders.py |
 | 2-1 | 이익의 질 | 빌더 있음, 미등록 | strategy/_income_builders.py |
 
@@ -312,7 +312,7 @@ c.review("수익구조")  # 특정 항목만
 - 사용자 제안: "재무제표 볼 때 자금조달 구조에서 시작 → 뭘 준비했고 → 돈을 어떻게 버는지"
 - 조사 결과: McKinsey ROIC + Penman Reformulation + DuPont 역순 읽기와 일치
 - 주류(Palepu-Healy, CFA, Buffett)는 사업 이해에서 시작 → Part 1 "사업을 먼저 본다" 유지
-- 1-1(수익 구조) 뒤에 1-2(자금) → 1-3(자산) → 1-4(ROIC) "자본의 여정" 흐름 삽입
+- 1-1(수익 구조) 뒤에 1-2(자금) → 1-3(자산) → 1-4(현금흐름) "자본의 여정" 흐름 삽입
 - Part 1이 5항목 → 7항목으로 확장
 
 ---
@@ -339,7 +339,7 @@ c.review("수익구조")  # 특정 항목만
 | 1-1 수익 구조 | strategy/ | **상세: `strategy/_01_revenueStructure.md`** |
 | 1-2 자금 구조 | strategy/ | 자본 구조(부채/자본 비중), 조달원 분석 |
 | 1-3 자산 구조 | strategy/ | 투하자본 구성, 유무형자산 비중, CAPEX 패턴 |
-| 1-4 ROIC | strategy/ | 투하자본수익률, McKinsey/Penman 기반 자본효율 |
+| 1-4 현금흐름 | strategy/ | CF 3구간, FCF, 이익의 현금 뒷받침, CF 패턴 |
 | 1-5 경쟁 우위 | strategy/ | 경쟁 포지셔닝, 진입장벽, 마진 지속성 |
 | 1-6 산업 사이클 | macro/ | gather/fred 활용, 산업 단계 판별 |
 | 1-7 지배구조/ESG | strategy/ | 현재 수준 유지 (esg/ 이미 있음) |

src/dartlab/analysis/strategy/cashflow.py

@@ -0,0 +1,237 @@
+"""1-4 현금흐름 — 계산만 담당.
+
+CF 3구간(영업/투자/재무) + FCF + 이익의 현금 뒷받침 + CF 패턴.
+블록 조립은 review/builders.py가 한다.
+"""
+
+from __future__ import annotations
+
+_MAX_YEARS = 5
+
+
+# ── 유틸 ──
+
+
+def _toDict(selectResult) -> tuple[dict[str, dict], list[str]] | None:
+    """SelectResult -> ({계정명: {period: val}}, periodCols)."""
+    from dartlab.analysis.strategy._helpers import toDict
+
+    return toDict(selectResult)
+
+
+def _annualCols(periods: list[str], maxYears: int = _MAX_YEARS) -> list[str]:
+    """연도 컬럼만 추출 (Q4 fallback)."""
+    cols = sorted([c for c in periods if "Q" not in c], reverse=True)
+    if cols:
+        return cols[:maxYears]
+    return sorted([c for c in periods if c.endswith("Q4")], reverse=True)[:maxYears]
+
+
+def _get(row: dict, col: str) -> float:
+    """dict에서 안전하게 값 꺼내기 (None -> 0)."""
+    v = row.get(col) if row else None
+    return v if v is not None else 0
+
+
+# ── CF 패턴 분류 ──
+
+
+def _classifyCfPattern(ocf: float, icf: float, fcf: float) -> str | None:
+    """영업/투자/재무 CF 부호 조합으로 패턴 분류."""
+
+    def _s(v: float) -> str:
+        if v > 0:
+            return "+"
+        if v < 0:
+            return "-"
+        return "0"
+
+    patterns = {
+        ("+", "-", "-"): "성숙형 — 영업으로 벌어 투자하고 부채 상환",
+        ("+", "-", "+"): "확장형 — 영업 + 외부 조달로 적극 투자",
+        ("+", "+", "-"): "구조조정형 — 자산 매각하며 부채 상환",
+        ("-", "-", "+"): "위기형 — 영업 적자를 외부 차입으로 메움",
+        ("-", "+", "+"): "축소형 — 자산 매각 + 차입으로 영업 적자 보전",
+        ("-", "+", "-"): "전환형 — 자산 매각으로 부채 상환, 영업 회복 필요",
+    }
+    return patterns.get((_s(ocf), _s(icf), _s(fcf)))
+
+
+# ── 메인: CF 3구간 + FCF ──
+
+
+def calcCashFlowOverview(company) -> dict | None:
+    """영업CF/투자CF/재무CF + FCF 시계열.
+
+    반환::
+
+        {
+            "history": [
+                {
+                    "period": str,
+                    "ocf": float, "icf": float, "fcf_financing": float,
+                    "capex": float, "fcf": float,
+                    "pattern": str | None,
+                },
+                ...
+            ],
+        }
+    """
+    cfAccounts = [
+        "영업활동현금흐름",
+        "투자활동현금흐름",
+        "재무활동으로인한현금흐름",
+        "유형자산의취득",
+        "무형자산의취득",
+    ]
+    result = company.select("CF", cfAccounts)
+    parsed = _toDict(result)
+    if parsed is None or "영업활동현금흐름" not in parsed[0]:
+        return None
+
+    data, allPeriods = parsed
+    ocfRow = data["영업활동현금흐름"]
+    icfRow = data.get("투자활동현금흐름", {})
+    finRow = data.get("재무활동으로인한현금흐름", {})
+    capexRow = data.get("유형자산의취득", {})
+    intCapexRow = data.get("무형자산의취득", {})
+
+    yCols = _annualCols(allPeriods, _MAX_YEARS)
+    if not yCols:
+        return None
+
+    history = []
+    for col in yCols:
+        ocf = _get(ocfRow, col)
+        icf = _get(icfRow, col)
+        fin = _get(finRow, col)
+        # CAPEX: CF에서 음수로 나옴 -> abs
+        capex = abs(_get(capexRow, col)) + abs(_get(intCapexRow, col))
+        fcf = ocf - capex
+
+        entry = {
+            "period": col,
+            "ocf": ocf,
+            "icf": icf,
+            "fcfFinancing": fin,
+            "capex": capex,
+            "fcf": fcf,
+            "pattern": _classifyCfPattern(ocf, icf, fin),
+        }
+        history.append(entry)
+
+    if not history:
+        return None
+    return {"history": history}
+
+
+# ── 이익의 현금 뒷받침 ──
+
+
+def calcCashQuality(company) -> dict | None:
+    """영업CF/순이익, 영업CF/매출 — 이익이 현금으로 뒷받침되는가.
+
+    반환::
+
+        {
+            "history": [
+                {
+                    "period": str,
+                    "ocf": float, "netIncome": float, "revenue": float,
+                    "ocfToNi": float | None,
+                    "ocfMargin": float | None,
+                },
+                ...
+            ],
+        }
+    """
+    cfResult = company.select("CF", ["영업활동현금흐름"])
+    isResult = company.select("IS", ["당기순이익", "매출액"])
+
+    cfParsed = _toDict(cfResult)
+    isParsed = _toDict(isResult)
+    if cfParsed is None or isParsed is None:
+        return None
+
+    cfData, cfPeriods = cfParsed
+    isData, _ = isParsed
+
+    ocfRow = cfData.get("영업활동현금흐름", {})
+    niRow = isData.get("당기순이익", {})
+    revRow = isData.get("매출액", {})
+
+    yCols = _annualCols(cfPeriods, _MAX_YEARS)
+    if not yCols:
+        return None
+
+    history = []
+    for col in yCols:
+        ocf = _get(ocfRow, col)
+        ni = _get(niRow, col)
+        rev = _get(revRow, col)
+
+        ocfToNi = ocf / ni * 100 if ni != 0 else None
+        ocfMargin = ocf / rev * 100 if rev > 0 else None
+
+        history.append(
+            {
+                "period": col,
+                "ocf": ocf,
+                "netIncome": ni,
+                "revenue": rev,
+                "ocfToNi": ocfToNi,
+                "ocfMargin": ocfMargin,
+            }
+        )
+
+    if not history:
+        return None
+    return {"history": history}
+
+
+# ── CF 플래그 ──
+
+
+def calcCashFlowFlags(company) -> list[str]:
+    """현금흐름 경고 신호."""
+    flags = []
+
+    overview = calcCashFlowOverview(company)
+    if overview and overview["history"]:
+        h0 = overview["history"][0]
+
+        # 영업CF 적자
+        if h0["ocf"] < 0:
+            flags.append("영업CF 적자 — 본업에서 현금이 나오지 않음")
+
+        # FCF 적자
+        if h0["fcf"] < 0 and h0["ocf"] > 0:
+            flags.append("FCF 적자 — 영업CF보다 투자가 큼")
+
+        # 위기형/축소형 패턴
+        pat = h0.get("pattern", "")
+        if pat and ("위기형" in pat or "축소형" in pat):
+            flags.append(f"CF 패턴: {pat}")
+
+        # 영업CF 3년 연속 감소
+        hist = overview["history"]
+        if len(hist) >= 3:
+            ocfs = [h["ocf"] for h in hist[:3]]
+            if ocfs[0] < ocfs[1] < ocfs[2]:
+                flags.append("영업CF 3년 연속 감소")
+
+    quality = calcCashQuality(company)
+    if quality and quality["history"]:
+        q0 = quality["history"][0]
+
+        # 영업CF/순이익 < 40% (이익 대비 현금 부족)
+        ratio = q0.get("ocfToNi")
+        if ratio is not None and 0 < ratio < 40:
+            flags.append(f"영업CF/순이익 {ratio:.0f}% — 이익의 현금 뒷받침 부족")
+
+        # 영업CF 마진 < 0
+        margin = q0.get("ocfMargin")
+        if margin is not None and margin < 0:
+            flags.append(f"영업CF 마진 {margin:.1f}% — 매출 대비 현금 유출")
+
+    return flags

src/dartlab/review/builders.py

@@ -803,3 +803,87 @@ def assetFlagsBlock(flags: list[str]) -> list:
     if not flags:
         return []
     return [FlagBlock(flags, kind="warning")]
+
+
+# ── 1-4 현금흐름 빌더 ──
+
+
+def cashFlowOverviewBlock(data: dict) -> list:
+    """calcCashFlowOverview 결과 → CF 3구간 + FCF 시계열 테이블."""
+    if not data:
+        return []
+    history = data.get("history", [])
+    if not history:
+        return []
+
+    blocks: list = []
+    blocks.append(
+        HeadingBlock(
+            "현금흐름 3구간",
+            level=2,
+            helper="영업CF(+)/투자CF(-)/재무CF(-) = 건전한 패턴",
+        )
+    )
+
+    rows = ["영업CF", "투자CF", "재무CF", "CAPEX", "FCF"]
+    cols = {"": rows}
+    for h in history:
+        cols[h["period"]] = [
+            _fmtAmtShort(h["ocf"]),
+            _fmtAmtShort(h["icf"]),
+            _fmtAmtShort(h["fcfFinancing"]),
+            _fmtAmtShort(h["capex"]),
+            _fmtAmtShort(h["fcf"]),
+        ]
+    blocks.append(TableBlock("현금흐름 추이", pl.DataFrame(cols)))
+
+    # CF 패턴 시계열
+    patternRows = ["CF 패턴"]
+    patternCols = {"": patternRows}
+    for h in history:
+        pat = h.get("pattern")
+        label = pat.split(" — ")[0] if pat else "-"
+        patternCols[h["period"]] = [label]
+    blocks.append(TableBlock("CF 패턴 추이", pl.DataFrame(patternCols)))
+
+    return blocks
+
+
+def cashQualityBlock(data: dict) -> list:
+    """calcCashQuality 결과 → 영업CF/순이익, 영업CF 마진 시계열."""
+    if not data:
+        return []
+    history = data.get("history", [])
+    if not history:
+        return []
+
+    blocks: list = []
+    blocks.append(
+        HeadingBlock(
+            "이익의 현금 뒷받침",
+            level=2,
+            helper="영업CF/순이익 > 100%이면 이익이 현금으로 회수됨",
+        )
+    )
+
+    rows = ["영업CF", "당기순이익", "영업CF/순이익", "영업CF 마진"]
+    cols = {"": rows}
+    for h in history:
+        ratio = h.get("ocfToNi")
+        margin = h.get("ocfMargin")
+        cols[h["period"]] = [
+            _fmtAmtShort(h["ocf"]),
+            _fmtAmtShort(h["netIncome"]),
+            f"{ratio:.0f}%" if ratio is not None else "-",
+            f"{margin:.1f}%" if margin is not None else "-",
+        ]
+    blocks.append(TableBlock("현금 품질 추이", pl.DataFrame(cols)))
+
+    return blocks
+
+
+def cashFlowFlagsBlock(flags: list[str]) -> list:
+    """calcCashFlowFlags 결과 → FlagBlock."""
+    if not flags:
+        return []
+    return [FlagBlock(flags, kind="warning")]