Financial Ratios Analysis in BI

Learning by Doing Series — Topic 8: Build Cash Flow Indicators

Objective

Create operating cash flow, capital expenditures, free cash flow, cash flow to debt, and cash flow coverage indicators.

The main output of this topic is Fact_Cash_Flow.csv.

Objetivo

Create operating cash flow, capital expenditures, free cash flow, cash flow to debt, and cash flow coverage indicators.

El archivo principal de salida de este tópico es Fact_Cash_Flow.csv.

Production / Cybersecurity Warning

This exercise uses synthetic training data only. Do not run scripts in production, real corporate folders, financial databases, or work environments without formal authorization. Validate backups, sandbox testing, least-privilege permissions, change-control approval, audit requirements, and organizational cybersecurity protocols before automating any real financial process.

Advertencia de Producción / Ciberseguridad

Este ejercicio usa data sintética. No ejecutes scripts en producción, carpetas reales, bases financieras corporativas o ambientes de trabajo sin autorización formal. Valida respaldos, pruebas en sandbox, permisos mínimos, control de cambios, auditoría y protocolos de ciberseguridad antes de automatizar cualquier proceso financiero real.

Business Scenario

The training has already created dimensions and core financial data. This topic moves one step closer to an executive BI model by creating a clean analytical file for Cash Flow.

Escenario de Negocio

El training ya creó dimensiones y data financiera base. Este tópico avanza hacia el modelo ejecutivo de BI creando un archivo analítico limpio para Cash Flow.

Input: Dim_Company.csv

Input: Dim_Period.csv

Output: Fact_Cash_Flow.csv

Focus: Cash Flow

Step-by-Step Practice

Step 1 — Prepare the folder and load previous files

Step 2 — Build topic-specific calculations

Step 3 — Export the output CSV or documentation file

Step 4 — Validate rows, keys, and financial logic

Step 5 — Interpret the result as a BI analyst

Script / Instructions

from pathlib import Path
import pandas as pd

# ============================================================
# Financial Ratios Analysis in BI
# Topic 8 - Build Cash Flow Indicators
# ============================================================

# ------------------------------------------------------------
# Step 1 - Set project folder
# ------------------------------------------------------------

base_path = Path("financial_ratios_bi_training")

if not base_path.exists():
    raise FileNotFoundError(
        "The project folder does not exist. Please run Topic 1 first."
    )

trial_balance_path = base_path / "Fact_Trial_Balance.csv"
mapping_path = base_path / "Mapping_Financial_Statements.csv"
balance_sheet_path = base_path / "Fact_Balance_Sheet.csv"
dim_company_path = base_path / "Dim_Company.csv"
dim_period_path = base_path / "Dim_Period.csv"

required_files = [
    trial_balance_path,
    mapping_path,
    balance_sheet_path,
    dim_company_path,
    dim_period_path
]

for file_path in required_files:
    if not file_path.exists():
        raise FileNotFoundError(
            f"Required file not found: {file_path.name}. "
            "Please run the previous topics first."
        )

print(f"Project folder found: {base_path.resolve()}")

# ------------------------------------------------------------
# Step 2 - Load source files
# ------------------------------------------------------------

trial_balance = pd.read_csv(trial_balance_path)
mapping = pd.read_csv(mapping_path)
balance_sheet = pd.read_csv(balance_sheet_path)
dim_company = pd.read_csv(dim_company_path)
dim_period = pd.read_csv(dim_period_path)

trial_balance["AccountNumber"] = trial_balance["AccountNumber"].astype(str)
mapping["AccountNumber"] = mapping["AccountNumber"].astype(str)

print("Fact_Trial_Balance loaded:", len(trial_balance), "rows")
print("Mapping_Financial_Statements loaded:", len(mapping), "rows")
print("Fact_Balance_Sheet loaded:", len(balance_sheet), "rows")

# ------------------------------------------------------------
# Step 3 - Filter Cash Flow Statement accounts
# ------------------------------------------------------------

cash_flow_mapping = mapping[
    mapping["FinancialStatement"] == "Cash Flow Statement"
].copy()

cash_flow_source = trial_balance.merge(
    cash_flow_mapping[
        [
            "AccountNumber",
            "StatementSection",
            "StatementLine",
            "LineGroup",
            "LineOrder",
            "SignMultiplier",
            "RatioInput"
        ]
    ],
    on="AccountNumber",
    how="inner"
)

print("Cash flow source rows:", len(cash_flow_source))

# ------------------------------------------------------------
# Step 4 - Calculate cash flow amount
# ------------------------------------------------------------
# In this training model:
# - Operating Cash Flow is shown as positive.
# - Capital Expenditures are shown as negative because they are cash outflows.
# - Dividends Paid are shown as negative because they are cash outflows.

cash_flow_source["RawAmount"] = cash_flow_source.apply(
    lambda row: row["Debit"] if row["Debit"] > 0 else row["Credit"],
    axis=1
)

cash_flow_source["Amount"] = cash_flow_source["RawAmount"].round(2)

cash_flow_source.loc[
    cash_flow_source["RatioInput"] == "Capital Expenditures",
    "Amount"
] = cash_flow_source.loc[
    cash_flow_source["RatioInput"] == "Capital Expenditures",
    "Amount"
] * -1

cash_flow_source.loc[
    cash_flow_source["RatioInput"] == "Dividends Paid",
    "Amount"
] = cash_flow_source.loc[
    cash_flow_source["RatioInput"] == "Dividends Paid",
    "Amount"
] * -1

cash_flow_source["Amount"] = cash_flow_source["Amount"].round(2)

# ------------------------------------------------------------
# Step 5 - Aggregate base cash flow lines
# ------------------------------------------------------------

base_lines = (
    cash_flow_source
    .groupby(
        [
            "CompanyID",
            "PeriodID",
            "StatementSection",
            "StatementLine",
            "LineGroup",
            "LineOrder",
            "RatioInput"
        ]
    )
    .agg(Amount=("Amount", "sum"))
    .reset_index()
)

base_lines["Amount"] = base_lines["Amount"].round(2)
base_lines["LineType"] = "Base Line"

# ------------------------------------------------------------
# Step 6 - Prepare debt values from Balance Sheet
# ------------------------------------------------------------

debt_lines = balance_sheet[
    balance_sheet["RatioInput"].isin(
        [
            "Short-Term Debt",
            "Long-Term Debt",
            "Total Liabilities"
        ]
    )
].copy()

debt_pivot = (
    debt_lines
    .pivot_table(
        index=["CompanyID", "PeriodID"],
        columns="RatioInput",
        values="Amount",
        aggfunc="sum"
    )
    .reset_index()
)

for col in ["Short-Term Debt", "Long-Term Debt", "Total Liabilities"]:
    if col not in debt_pivot.columns:
        debt_pivot[col] = 0

debt_pivot["TotalDebt"] = (
    debt_pivot["Short-Term Debt"]
    + debt_pivot["Long-Term Debt"]
).round(2)

# ------------------------------------------------------------
# Step 7 - Create calculated cash flow indicators
# ------------------------------------------------------------

calculated_records = []

for (company_id, period_id), group in base_lines.groupby(["CompanyID", "PeriodID"]):

    operating_cash_flow = group.loc[
        group["RatioInput"] == "Operating Cash Flow",
        "Amount"
    ].sum()

    capital_expenditures = group.loc[
        group["RatioInput"] == "Capital Expenditures",
        "Amount"
    ].sum()

    dividends_paid = group.loc[
        group["RatioInput"] == "Dividends Paid",
        "Amount"
    ].sum()

    free_cash_flow = operating_cash_flow + capital_expenditures
    net_cash_flow_after_dividends = free_cash_flow + dividends_paid

    debt_row = debt_pivot[
        (debt_pivot["CompanyID"] == company_id)
        & (debt_pivot["PeriodID"] == period_id)
    ]

    if debt_row.empty:
        total_debt = 0
        total_liabilities = 0
    else:
        total_debt = float(debt_row["TotalDebt"].iloc[0])
        total_liabilities = float(debt_row["Total Liabilities"].iloc[0])

    cash_flow_to_debt = (
        operating_cash_flow / total_debt
        if total_debt != 0
        else 0
    )

    cash_flow_coverage = (
        operating_cash_flow / total_liabilities
        if total_liabilities != 0
        else 0
    )

    calculated_records.extend([
        {
            "CompanyID": company_id,
            "PeriodID": period_id,
            "StatementSection": "Cash Flow Indicators",
            "StatementLine": "Free Cash Flow",
            "LineGroup": "Calculated Cash Flow Lines",
            "LineOrder": 1250,
            "RatioInput": "Free Cash Flow",
            "Amount": round(free_cash_flow, 2),
            "LineType": "Calculated Line"
        },
        {
            "CompanyID": company_id,
            "PeriodID": period_id,
            "StatementSection": "Cash Flow Indicators",
            "StatementLine": "Net Cash Flow After Dividends",
            "LineGroup": "Calculated Cash Flow Lines",
            "LineOrder": 1350,
            "RatioInput": "Net Cash Flow After Dividends",
            "Amount": round(net_cash_flow_after_dividends, 2),
            "LineType": "Calculated Line"
        },
        {
            "CompanyID": company_id,
            "PeriodID": period_id,
            "StatementSection": "Cash Flow Indicators",
            "StatementLine": "Cash Flow to Debt",
            "LineGroup": "Calculated Cash Flow Ratios",
            "LineOrder": 1400,
            "RatioInput": "Cash Flow to Debt",
            "Amount": round(cash_flow_to_debt, 4),
            "LineType": "Calculated Ratio"
        },
        {
            "CompanyID": company_id,
            "PeriodID": period_id,
            "StatementSection": "Cash Flow Indicators",
            "StatementLine": "Cash Flow Coverage",
            "LineGroup": "Calculated Cash Flow Ratios",
            "LineOrder": 1410,
            "RatioInput": "Cash Flow Coverage",
            "Amount": round(cash_flow_coverage, 4),
            "LineType": "Calculated Ratio"
        }
    ])

calculated_lines = pd.DataFrame(calculated_records)

# ------------------------------------------------------------
# Step 8 - Combine base and calculated lines
# ------------------------------------------------------------

fact_cash_flow = pd.concat(
    [
        base_lines,
        calculated_lines
    ],
    ignore_index=True
)

fact_cash_flow = fact_cash_flow.sort_values(
    [
        "CompanyID",
        "PeriodID",
        "LineOrder",
        "StatementLine"
    ]
).reset_index(drop=True)

# ------------------------------------------------------------
# Step 9 - Add company and period labels
# ------------------------------------------------------------

fact_cash_flow = fact_cash_flow.merge(
    dim_company[["CompanyID", "CompanyName", "Industry"]],
    on="CompanyID",
    how="left"
)

fact_cash_flow = fact_cash_flow.merge(
    dim_period[
        [
            "PeriodID",
            "FiscalYear",
            "FiscalQuarter",
            "YearQuarter",
            "PeriodEndDate"
        ]
    ],
    on="PeriodID",
    how="left"
)

fact_cash_flow = fact_cash_flow[
    [
        "CompanyID",
        "CompanyName",
        "Industry",
        "PeriodID",
        "FiscalYear",
        "FiscalQuarter",
        "YearQuarter",
        "PeriodEndDate",
        "StatementSection",
        "StatementLine",
        "LineGroup",
        "LineOrder",
        "RatioInput",
        "LineType",
        "Amount"
    ]
]

# ------------------------------------------------------------
# Step 10 - Export Fact_Cash_Flow
# ------------------------------------------------------------

fact_cash_flow.to_csv(
    base_path / "Fact_Cash_Flow.csv",
    index=False
)

print("Fact_Cash_Flow.csv created successfully.")

# ------------------------------------------------------------
# Step 11 - Create cash flow validation summary
# ------------------------------------------------------------

validation_records = []

for (company_id, period_id), group in fact_cash_flow.groupby(["CompanyID", "PeriodID"]):

    operating_cash_flow = group.loc[
        group["RatioInput"] == "Operating Cash Flow",
        "Amount"
    ].sum()

    capital_expenditures = group.loc[
        group["RatioInput"] == "Capital Expenditures",
        "Amount"
    ].sum()

    free_cash_flow = group.loc[
        group["RatioInput"] == "Free Cash Flow",
        "Amount"
    ].sum()

    dividends_paid = group.loc[
        group["RatioInput"] == "Dividends Paid",
        "Amount"
    ].sum()

    net_cash_flow_after_dividends = group.loc[
        group["RatioInput"] == "Net Cash Flow After Dividends",
        "Amount"
    ].sum()

    expected_free_cash_flow = operating_cash_flow + capital_expenditures
    free_cash_flow_difference = round(
        free_cash_flow - expected_free_cash_flow,
        2
    )

    expected_net_cash_flow = free_cash_flow + dividends_paid
    net_cash_flow_difference = round(
        net_cash_flow_after_dividends - expected_net_cash_flow,
        2
    )

    validation_records.append({
        "CompanyID": company_id,
        "PeriodID": period_id,
        "OperatingCashFlow": round(operating_cash_flow, 2),
        "CapitalExpenditures": round(capital_expenditures, 2),
        "FreeCashFlow": round(free_cash_flow, 2),
        "ExpectedFreeCashFlow": round(expected_free_cash_flow, 2),
        "FreeCashFlowDifference": free_cash_flow_difference,
        "DividendsPaid": round(dividends_paid, 2),
        "NetCashFlowAfterDividends": round(net_cash_flow_after_dividends, 2),
        "ExpectedNetCashFlowAfterDividends": round(expected_net_cash_flow, 2),
        "NetCashFlowDifference": net_cash_flow_difference,
        "Status": (
            "PASSED"
            if free_cash_flow_difference == 0
            and net_cash_flow_difference == 0
            else "FAILED"
        )
    })

cash_flow_validation = pd.DataFrame(validation_records)

cash_flow_validation.to_csv(
    base_path / "Cash_Flow_Validation_Summary.csv",
    index=False
)

print("Cash_Flow_Validation_Summary.csv created successfully.")

# ------------------------------------------------------------
# Step 12 - Print validation summary
# ------------------------------------------------------------

print()
print("==============================")
print("VALIDATION SUMMARY")
print("==============================")

print("Cash flow rows:", len(fact_cash_flow))
print("Validation rows:", len(cash_flow_validation))

print()
print("Rows by LineType:")
print(
    fact_cash_flow
    .groupby("LineType")["StatementLine"]
    .count()
    .reset_index(name="NumberOfRows")
)

print()
print("Cash Flow Lines:")
print(
    fact_cash_flow[
        [
            "LineOrder",
            "StatementLine",
            "LineType"
        ]
    ]
    .drop_duplicates()
    .sort_values("LineOrder")
)

print()
print("Cash Flow Validation Summary:")
print(cash_flow_validation.head(20))

print()
print("Failed validation rows:")
failed_rows = cash_flow_validation[cash_flow_validation["Status"] == "FAILED"]

if failed_rows.empty:
    print("No failed rows. Cash Flow validation passed.")
else:
    print(failed_rows)

print()
print("Files currently in project folder:")

for file in base_path.glob("*.csv"):
    print("-", file.name)

print()
print("Topic 8 completed successfully.")

Expected Output

The script or instructions create:

Resultado Esperado

El script o las instrucciones crean:

financial_ratios_bi_training/Fact_Cash_Flow.csv

Validation Checklist

Checkpoint 1: Validate source fields, row counts, relationship keys, financial reasonableness, and business interpretation before moving to the next topic.
Checkpoint 2: Validate source fields, row counts, relationship keys, financial reasonableness, and business interpretation before moving to the next topic.
Checkpoint 3: Validate source fields, row counts, relationship keys, financial reasonableness, and business interpretation before moving to the next topic.
Checkpoint 4: Validate source fields, row counts, relationship keys, financial reasonableness, and business interpretation before moving to the next topic.
Checkpoint 5: Validate source fields, row counts, relationship keys, financial reasonableness, and business interpretation before moving to the next topic.
Checkpoint 6: Validate source fields, row counts, relationship keys, financial reasonableness, and business interpretation before moving to the next topic.
Checkpoint 7: Validate source fields, row counts, relationship keys, financial reasonableness, and business interpretation before moving to the next topic.
Checkpoint 8: Validate source fields, row counts, relationship keys, financial reasonableness, and business interpretation before moving to the next topic.

Business Interpretation

This topic is not only a technical exercise. It teaches students how to translate accounting data into management information. The output becomes part of the BI layer used later for ratios, dashboards, trend analysis, benchmarking, and executive decision-making.

Interpretación de Negocio

Este tópico no es solo un ejercicio técnico. Enseña a convertir data contable en información gerencial. El archivo de salida se integra a la capa de BI que luego se usa para ratios, dashboards, tendencias, comparación entre compañías y decisiones ejecutivas.

Final Recommendation

Before moving forward, open the generated file, review the columns, confirm the number of companies and periods, and make sure the numbers make financial sense. Good BI starts with clean, explainable, and validated data.

Final Validated Script

This is the corrected and live-tested script for Topic 8: Build Cash Flow Indicators.

from pathlib import Path
import pandas as pd

# ============================================================
# Financial Ratios Analysis in BI
# Topic 8 - Build Cash Flow Indicators
# ============================================================

# ------------------------------------------------------------
# Step 1 - Set project folder
# ------------------------------------------------------------

base_path = Path("financial_ratios_bi_training")

if not base_path.exists():
    raise FileNotFoundError(
        "The project folder does not exist. Please run Topic 1 first."
    )

trial_balance_path = base_path / "Fact_Trial_Balance.csv"
mapping_path = base_path / "Mapping_Financial_Statements.csv"
balance_sheet_path = base_path / "Fact_Balance_Sheet.csv"
dim_company_path = base_path / "Dim_Company.csv"
dim_period_path = base_path / "Dim_Period.csv"

required_files = [
    trial_balance_path,
    mapping_path,
    balance_sheet_path,
    dim_company_path,
    dim_period_path
]

for file_path in required_files:
    if not file_path.exists():
        raise FileNotFoundError(
            f"Required file not found: {file_path.name}. "
            "Please run the previous topics first."
        )

print(f"Project folder found: {base_path.resolve()}")

# ------------------------------------------------------------
# Step 2 - Load source files
# ------------------------------------------------------------

trial_balance = pd.read_csv(trial_balance_path)
mapping = pd.read_csv(mapping_path)
balance_sheet = pd.read_csv(balance_sheet_path)
dim_company = pd.read_csv(dim_company_path)
dim_period = pd.read_csv(dim_period_path)

trial_balance["AccountNumber"] = trial_balance["AccountNumber"].astype(str)
mapping["AccountNumber"] = mapping["AccountNumber"].astype(str)

print("Fact_Trial_Balance loaded:", len(trial_balance), "rows")
print("Mapping_Financial_Statements loaded:", len(mapping), "rows")
print("Fact_Balance_Sheet loaded:", len(balance_sheet), "rows")

# ------------------------------------------------------------
# Step 3 - Filter Cash Flow Statement accounts
# ------------------------------------------------------------

cash_flow_mapping = mapping[
    mapping["FinancialStatement"] == "Cash Flow Statement"
].copy()

cash_flow_source = trial_balance.merge(
    cash_flow_mapping[
        [
            "AccountNumber",
            "StatementSection",
            "StatementLine",
            "LineGroup",
            "LineOrder",
            "SignMultiplier",
            "RatioInput"
        ]
    ],
    on="AccountNumber",
    how="inner"
)

print("Cash flow source rows:", len(cash_flow_source))

# ------------------------------------------------------------
# Step 4 - Calculate cash flow amount
# ------------------------------------------------------------
# In this training model:
# - Operating Cash Flow is shown as positive.
# - Capital Expenditures are shown as negative because they are cash outflows.
# - Dividends Paid are shown as negative because they are cash outflows.

cash_flow_source["RawAmount"] = cash_flow_source.apply(
    lambda row: row["Debit"] if row["Debit"] > 0 else row["Credit"],
    axis=1
)

cash_flow_source["Amount"] = cash_flow_source["RawAmount"].round(2)

cash_flow_source.loc[
    cash_flow_source["RatioInput"] == "Capital Expenditures",
    "Amount"
] = cash_flow_source.loc[
    cash_flow_source["RatioInput"] == "Capital Expenditures",
    "Amount"
] * -1

cash_flow_source.loc[
    cash_flow_source["RatioInput"] == "Dividends Paid",
    "Amount"
] = cash_flow_source.loc[
    cash_flow_source["RatioInput"] == "Dividends Paid",
    "Amount"
] * -1

cash_flow_source["Amount"] = cash_flow_source["Amount"].round(2)

# ------------------------------------------------------------
# Step 5 - Aggregate base cash flow lines
# ------------------------------------------------------------

base_lines = (
    cash_flow_source
    .groupby(
        [
            "CompanyID",
            "PeriodID",
            "StatementSection",
            "StatementLine",
            "LineGroup",
            "LineOrder",
            "RatioInput"
        ]
    )
    .agg(Amount=("Amount", "sum"))
    .reset_index()
)

base_lines["Amount"] = base_lines["Amount"].round(2)
base_lines["LineType"] = "Base Line"

# ------------------------------------------------------------
# Step 6 - Prepare debt values from Balance Sheet
# ------------------------------------------------------------

debt_lines = balance_sheet[
    balance_sheet["RatioInput"].isin(
        [
            "Short-Term Debt",
            "Long-Term Debt",
            "Total Liabilities"
        ]
    )
].copy()

debt_pivot = (
    debt_lines
    .pivot_table(
        index=["CompanyID", "PeriodID"],
        columns="RatioInput",
        values="Amount",
        aggfunc="sum"
    )
    .reset_index()
)

for col in ["Short-Term Debt", "Long-Term Debt", "Total Liabilities"]:
    if col not in debt_pivot.columns:
        debt_pivot[col] = 0

debt_pivot["TotalDebt"] = (
    debt_pivot["Short-Term Debt"]
    + debt_pivot["Long-Term Debt"]
).round(2)

# ------------------------------------------------------------
# Step 7 - Create calculated cash flow indicators
# ------------------------------------------------------------

calculated_records = []

for (company_id, period_id), group in base_lines.groupby(["CompanyID", "PeriodID"]):

    operating_cash_flow = group.loc[
        group["RatioInput"] == "Operating Cash Flow",
        "Amount"
    ].sum()

    capital_expenditures = group.loc[
        group["RatioInput"] == "Capital Expenditures",
        "Amount"
    ].sum()

    dividends_paid = group.loc[
        group["RatioInput"] == "Dividends Paid",
        "Amount"
    ].sum()

    free_cash_flow = operating_cash_flow + capital_expenditures
    net_cash_flow_after_dividends = free_cash_flow + dividends_paid

    debt_row = debt_pivot[
        (debt_pivot["CompanyID"] == company_id)
        & (debt_pivot["PeriodID"] == period_id)
    ]

    if debt_row.empty:
        total_debt = 0
        total_liabilities = 0
    else:
        total_debt = float(debt_row["TotalDebt"].iloc[0])
        total_liabilities = float(debt_row["Total Liabilities"].iloc[0])

    cash_flow_to_debt = (
        operating_cash_flow / total_debt
        if total_debt != 0
        else 0
    )

    cash_flow_coverage = (
        operating_cash_flow / total_liabilities
        if total_liabilities != 0
        else 0
    )

    calculated_records.extend([
        {
            "CompanyID": company_id,
            "PeriodID": period_id,
            "StatementSection": "Cash Flow Indicators",
            "StatementLine": "Free Cash Flow",
            "LineGroup": "Calculated Cash Flow Lines",
            "LineOrder": 1250,
            "RatioInput": "Free Cash Flow",
            "Amount": round(free_cash_flow, 2),
            "LineType": "Calculated Line"
        },
        {
            "CompanyID": company_id,
            "PeriodID": period_id,
            "StatementSection": "Cash Flow Indicators",
            "StatementLine": "Net Cash Flow After Dividends",
            "LineGroup": "Calculated Cash Flow Lines",
            "LineOrder": 1350,
            "RatioInput": "Net Cash Flow After Dividends",
            "Amount": round(net_cash_flow_after_dividends, 2),
            "LineType": "Calculated Line"
        },
        {
            "CompanyID": company_id,
            "PeriodID": period_id,
            "StatementSection": "Cash Flow Indicators",
            "StatementLine": "Cash Flow to Debt",
            "LineGroup": "Calculated Cash Flow Ratios",
            "LineOrder": 1400,
            "RatioInput": "Cash Flow to Debt",
            "Amount": round(cash_flow_to_debt, 4),
            "LineType": "Calculated Ratio"
        },
        {
            "CompanyID": company_id,
            "PeriodID": period_id,
            "StatementSection": "Cash Flow Indicators",
            "StatementLine": "Cash Flow Coverage",
            "LineGroup": "Calculated Cash Flow Ratios",
            "LineOrder": 1410,
            "RatioInput": "Cash Flow Coverage",
            "Amount": round(cash_flow_coverage, 4),
            "LineType": "Calculated Ratio"
        }
    ])

calculated_lines = pd.DataFrame(calculated_records)

# ------------------------------------------------------------
# Step 8 - Combine base and calculated lines
# ------------------------------------------------------------

fact_cash_flow = pd.concat(
    [
        base_lines,
        calculated_lines
    ],
    ignore_index=True
)

fact_cash_flow = fact_cash_flow.sort_values(
    [
        "CompanyID",
        "PeriodID",
        "LineOrder",
        "StatementLine"
    ]
).reset_index(drop=True)

# ------------------------------------------------------------
# Step 9 - Add company and period labels
# ------------------------------------------------------------

fact_cash_flow = fact_cash_flow.merge(
    dim_company[["CompanyID", "CompanyName", "Industry"]],
    on="CompanyID",
    how="left"
)

fact_cash_flow = fact_cash_flow.merge(
    dim_period[
        [
            "PeriodID",
            "FiscalYear",
            "FiscalQuarter",
            "YearQuarter",
            "PeriodEndDate"
        ]
    ],
    on="PeriodID",
    how="left"
)

fact_cash_flow = fact_cash_flow[
    [
        "CompanyID",
        "CompanyName",
        "Industry",
        "PeriodID",
        "FiscalYear",
        "FiscalQuarter",
        "YearQuarter",
        "PeriodEndDate",
        "StatementSection",
        "StatementLine",
        "LineGroup",
        "LineOrder",
        "RatioInput",
        "LineType",
        "Amount"
    ]
]

# ------------------------------------------------------------
# Step 10 - Export Fact_Cash_Flow
# ------------------------------------------------------------

fact_cash_flow.to_csv(
    base_path / "Fact_Cash_Flow.csv",
    index=False
)

print("Fact_Cash_Flow.csv created successfully.")

# ------------------------------------------------------------
# Step 11 - Create cash flow validation summary
# ------------------------------------------------------------

validation_records = []

for (company_id, period_id), group in fact_cash_flow.groupby(["CompanyID", "PeriodID"]):

    operating_cash_flow = group.loc[
        group["RatioInput"] == "Operating Cash Flow",
        "Amount"
    ].sum()

    capital_expenditures = group.loc[
        group["RatioInput"] == "Capital Expenditures",
        "Amount"
    ].sum()

    free_cash_flow = group.loc[
        group["RatioInput"] == "Free Cash Flow",
        "Amount"
    ].sum()

    dividends_paid = group.loc[
        group["RatioInput"] == "Dividends Paid",
        "Amount"
    ].sum()

    net_cash_flow_after_dividends = group.loc[
        group["RatioInput"] == "Net Cash Flow After Dividends",
        "Amount"
    ].sum()

    expected_free_cash_flow = operating_cash_flow + capital_expenditures
    free_cash_flow_difference = round(
        free_cash_flow - expected_free_cash_flow,
        2
    )

    expected_net_cash_flow = free_cash_flow + dividends_paid
    net_cash_flow_difference = round(
        net_cash_flow_after_dividends - expected_net_cash_flow,
        2
    )

    validation_records.append({
        "CompanyID": company_id,
        "PeriodID": period_id,
        "OperatingCashFlow": round(operating_cash_flow, 2),
        "CapitalExpenditures": round(capital_expenditures, 2),
        "FreeCashFlow": round(free_cash_flow, 2),
        "ExpectedFreeCashFlow": round(expected_free_cash_flow, 2),
        "FreeCashFlowDifference": free_cash_flow_difference,
        "DividendsPaid": round(dividends_paid, 2),
        "NetCashFlowAfterDividends": round(net_cash_flow_after_dividends, 2),
        "ExpectedNetCashFlowAfterDividends": round(expected_net_cash_flow, 2),
        "NetCashFlowDifference": net_cash_flow_difference,
        "Status": (
            "PASSED"
            if free_cash_flow_difference == 0
            and net_cash_flow_difference == 0
            else "FAILED"
        )
    })

cash_flow_validation = pd.DataFrame(validation_records)

cash_flow_validation.to_csv(
    base_path / "Cash_Flow_Validation_Summary.csv",
    index=False
)

print("Cash_Flow_Validation_Summary.csv created successfully.")

# ------------------------------------------------------------
# Step 12 - Print validation summary
# ------------------------------------------------------------

print()
print("==============================")
print("VALIDATION SUMMARY")
print("==============================")

print("Cash flow rows:", len(fact_cash_flow))
print("Validation rows:", len(cash_flow_validation))

print()
print("Rows by LineType:")
print(
    fact_cash_flow
    .groupby("LineType")["StatementLine"]
    .count()
    .reset_index(name="NumberOfRows")
)

print()
print("Cash Flow Lines:")
print(
    fact_cash_flow[
        [
            "LineOrder",
            "StatementLine",
            "LineType"
        ]
    ]
    .drop_duplicates()
    .sort_values("LineOrder")
)

print()
print("Cash Flow Validation Summary:")
print(cash_flow_validation.head(20))

print()
print("Failed validation rows:")
failed_rows = cash_flow_validation[cash_flow_validation["Status"] == "FAILED"]

if failed_rows.empty:
    print("No failed rows. Cash Flow validation passed.")
else:
    print(failed_rows)

print()
print("Files currently in project folder:")

for file in base_path.glob("*.csv"):
    print("-", file.name)

print()
print("Topic 8 completed successfully.")