Financial Ratios Analysis in BI

Learning by Doing Series — Topic 14: Cash Flow Indicator Ratios

Objective

Calculate operating cash flow to sales, free cash flow to operating cash flow, dividend payout ratio, and cash flow coverage ratio.

The main output of this topic is Fact_Cash_Flow_Ratios.csv.

Objetivo

Calculate operating cash flow to sales, free cash flow to operating cash flow, dividend payout ratio, and cash flow coverage ratio.

El archivo principal de salida de este tópico es Fact_Cash_Flow_Ratios.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 Ratios.

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 Ratios.

Input: Dim_Company.csv
Input: Dim_Period.csv
Output: Fact_Cash_Flow_Ratios.csv
Focus: Cash Flow Ratios

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 14 - Cash Flow Indicator Ratios
# ============================================================

# ------------------------------------------------------------
# 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."
    )

cash_flow_path = base_path / "Fact_Cash_Flow.csv"
income_statement_path = base_path / "Fact_Income_Statement.csv"
balance_sheet_path = base_path / "Fact_Balance_Sheet.csv"
market_data_path = base_path / "Fact_Market_Data.csv"
dim_company_path = base_path / "Dim_Company.csv"
dim_period_path = base_path / "Dim_Period.csv"

required_files = [
    cash_flow_path,
    income_statement_path,
    balance_sheet_path,
    market_data_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
# ------------------------------------------------------------

cash_flow = pd.read_csv(cash_flow_path)
income_statement = pd.read_csv(income_statement_path)
balance_sheet = pd.read_csv(balance_sheet_path)
market_data = pd.read_csv(market_data_path)
dim_company = pd.read_csv(dim_company_path)
dim_period = pd.read_csv(dim_period_path)

print("Fact_Cash_Flow loaded:", len(cash_flow), "rows")
print("Fact_Income_Statement loaded:", len(income_statement), "rows")
print("Fact_Balance_Sheet loaded:", len(balance_sheet), "rows")
print("Fact_Market_Data loaded:", len(market_data), "rows")

# ------------------------------------------------------------
# Step 3 - Extract cash flow inputs
# ------------------------------------------------------------

cash_flow_inputs = cash_flow[
    cash_flow["RatioInput"].isin(
        [
            "Operating Cash Flow",
            "Capital Expenditures",
            "Free Cash Flow",
            "Dividends Paid",
            "Net Cash Flow After Dividends",
            "Cash Flow to Debt",
            "Cash Flow Coverage"
        ]
    )
].copy()

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

required_cash_flow_columns = [
    "Operating Cash Flow",
    "Capital Expenditures",
    "Free Cash Flow",
    "Dividends Paid",
    "Net Cash Flow After Dividends",
    "Cash Flow to Debt",
    "Cash Flow Coverage"
]

for column in required_cash_flow_columns:
    if column not in cash_flow_pivot.columns:
        cash_flow_pivot[column] = 0

# ------------------------------------------------------------
# Step 4 - Extract income statement inputs
# ------------------------------------------------------------

income_inputs = income_statement[
    income_statement["RatioInput"].isin(
        [
            "Revenue",
            "Net Income"
        ]
    )
].copy()

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

for column in ["Revenue", "Net Income"]:
    if column not in income_pivot.columns:
        income_pivot[column] = 0

# ------------------------------------------------------------
# Step 5 - Extract balance sheet inputs
# ------------------------------------------------------------

balance_inputs = balance_sheet[
    balance_sheet["RatioInput"].isin(
        [
            "Total Liabilities",
            "Total Assets",
            "Total Equity"
        ]
    )
].copy()

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

for column in ["Total Liabilities", "Total Assets", "Total Equity"]:
    if column not in balance_pivot.columns:
        balance_pivot[column] = 0

# ------------------------------------------------------------
# Step 6 - Extract market inputs
# ------------------------------------------------------------

market_inputs = market_data[
    [
        "CompanyID",
        "PeriodID",
        "DividendsPaid",
        "MarketCapitalization",
        "EnterpriseValue"
    ]
].copy()

# ------------------------------------------------------------
# Step 7 - Create base table
# ------------------------------------------------------------

cash_flow_ratio_base = (
    cash_flow_pivot
    .merge(income_pivot, on=["CompanyID", "PeriodID"], how="left")
    .merge(balance_pivot, on=["CompanyID", "PeriodID"], how="left")
    .merge(market_inputs, on=["CompanyID", "PeriodID"], how="left")
)

cash_flow_ratio_base = cash_flow_ratio_base.fillna(0)

# ------------------------------------------------------------
# Step 8 - Calculate cash flow ratios
# ------------------------------------------------------------

records = []

for _, row in cash_flow_ratio_base.iterrows():

    company_id = int(row["CompanyID"])
    period_id = int(row["PeriodID"])

    operating_cash_flow = float(row["Operating Cash Flow"])
    capital_expenditures = float(row["Capital Expenditures"])
    free_cash_flow = float(row["Free Cash Flow"])
    dividends_paid_cash_flow = abs(float(row["Dividends Paid"]))
    net_cash_flow_after_dividends = float(row["Net Cash Flow After Dividends"])

    revenue = float(row["Revenue"])
    net_income = float(row["Net Income"])

    total_liabilities = float(row["Total Liabilities"])
    total_assets = float(row["Total Assets"])
    total_equity = float(row["Total Equity"])

    dividends_paid_market = float(row["DividendsPaid"])
    market_cap = float(row["MarketCapitalization"])
    enterprise_value = float(row["EnterpriseValue"])

    operating_cash_flow_to_sales = (
        operating_cash_flow / revenue
        if revenue != 0
        else 0
    )

    free_cash_flow_to_operating_cash_flow = (
        free_cash_flow / operating_cash_flow
        if operating_cash_flow != 0
        else 0
    )

    free_cash_flow_margin = (
        free_cash_flow / revenue
        if revenue != 0
        else 0
    )

    operating_cash_flow_margin = (
        operating_cash_flow / revenue
        if revenue != 0
        else 0
    )

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

    dividend_payout_ratio = (
        dividends_paid_market / net_income
        if net_income != 0
        else 0
    )

    cash_return_on_assets = (
        operating_cash_flow / total_assets
        if total_assets != 0
        else 0
    )

    cash_return_on_equity = (
        operating_cash_flow / total_equity
        if total_equity != 0
        else 0
    )

    free_cash_flow_yield = (
        free_cash_flow / market_cap
        if market_cap != 0
        else 0
    )

    cash_flow_to_enterprise_value = (
        operating_cash_flow / enterprise_value
        if enterprise_value != 0
        else 0
    )

    records.extend([
        {
            "CompanyID": company_id,
            "PeriodID": period_id,
            "RatioCategory": "Cash Flow",
            "RatioName": "Operating Cash Flow",
            "RatioValue": round(operating_cash_flow, 2),
            "RatioFormat": "Currency",
            "Interpretation": "Cash generated from core operating activities."
        },
        {
            "CompanyID": company_id,
            "PeriodID": period_id,
            "RatioCategory": "Cash Flow",
            "RatioName": "Free Cash Flow",
            "RatioValue": round(free_cash_flow, 2),
            "RatioFormat": "Currency",
            "Interpretation": "Operating cash flow minus capital expenditures."
        },
        {
            "CompanyID": company_id,
            "PeriodID": period_id,
            "RatioCategory": "Cash Flow",
            "RatioName": "Operating Cash Flow to Sales",
            "RatioValue": round(operating_cash_flow_to_sales, 4),
            "RatioFormat": "Percent",
            "Interpretation": "Measures operating cash flow generated from each dollar of sales."
        },
        {
            "CompanyID": company_id,
            "PeriodID": period_id,
            "RatioCategory": "Cash Flow",
            "RatioName": "Free Cash Flow to Operating Cash Flow",
            "RatioValue": round(free_cash_flow_to_operating_cash_flow, 4),
            "RatioFormat": "Percent",
            "Interpretation": "Measures how much operating cash flow remains after capital expenditures."
        },
        {
            "CompanyID": company_id,
            "PeriodID": period_id,
            "RatioCategory": "Cash Flow",
            "RatioName": "Free Cash Flow Margin",
            "RatioValue": round(free_cash_flow_margin, 4),
            "RatioFormat": "Percent",
            "Interpretation": "Measures free cash flow generated as a percentage of revenue."
        },
        {
            "CompanyID": company_id,
            "PeriodID": period_id,
            "RatioCategory": "Cash Flow",
            "RatioName": "Operating Cash Flow Margin",
            "RatioValue": round(operating_cash_flow_margin, 4),
            "RatioFormat": "Percent",
            "Interpretation": "Measures operating cash flow as a percentage of revenue."
        },
        {
            "CompanyID": company_id,
            "PeriodID": period_id,
            "RatioCategory": "Cash Flow",
            "RatioName": "Cash Flow Coverage Ratio",
            "RatioValue": round(cash_flow_coverage_ratio, 4),
            "RatioFormat": "Decimal",
            "Interpretation": "Measures ability to cover total liabilities with operating cash flow."
        },
        {
            "CompanyID": company_id,
            "PeriodID": period_id,
            "RatioCategory": "Cash Flow",
            "RatioName": "Dividend Payout Ratio",
            "RatioValue": round(dividend_payout_ratio, 4),
            "RatioFormat": "Percent",
            "Interpretation": "Measures dividends paid as a percentage of net income."
        },
        {
            "CompanyID": company_id,
            "PeriodID": period_id,
            "RatioCategory": "Cash Flow",
            "RatioName": "Cash Return on Assets",
            "RatioValue": round(cash_return_on_assets, 4),
            "RatioFormat": "Percent",
            "Interpretation": "Measures operating cash flow generated by total assets."
        },
        {
            "CompanyID": company_id,
            "PeriodID": period_id,
            "RatioCategory": "Cash Flow",
            "RatioName": "Cash Return on Equity",
            "RatioValue": round(cash_return_on_equity, 4),
            "RatioFormat": "Percent",
            "Interpretation": "Measures operating cash flow generated by shareholder equity."
        },
        {
            "CompanyID": company_id,
            "PeriodID": period_id,
            "RatioCategory": "Cash Flow",
            "RatioName": "Free Cash Flow Yield",
            "RatioValue": round(free_cash_flow_yield, 4),
            "RatioFormat": "Percent",
            "Interpretation": "Measures free cash flow relative to market capitalization."
        },
        {
            "CompanyID": company_id,
            "PeriodID": period_id,
            "RatioCategory": "Cash Flow",
            "RatioName": "Cash Flow to Enterprise Value",
            "RatioValue": round(cash_flow_to_enterprise_value, 4),
            "RatioFormat": "Percent",
            "Interpretation": "Measures operating cash flow relative to enterprise value."
        }
    ])

fact_cash_flow_ratios = pd.DataFrame(records)

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

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

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

fact_cash_flow_ratios = fact_cash_flow_ratios[
    [
        "CompanyID",
        "CompanyName",
        "Industry",
        "PeriodID",
        "FiscalYear",
        "FiscalQuarter",
        "YearQuarter",
        "PeriodEndDate",
        "RatioCategory",
        "RatioName",
        "RatioValue",
        "RatioFormat",
        "Interpretation"
    ]
]

# ------------------------------------------------------------
# Step 10 - Export Fact_Cash_Flow_Ratios
# ------------------------------------------------------------

fact_cash_flow_ratios.to_csv(
    base_path / "Fact_Cash_Flow_Ratios.csv",
    index=False
)

print("Fact_Cash_Flow_Ratios.csv created successfully.")

# ------------------------------------------------------------
# Step 11 - Create validation summary
# ------------------------------------------------------------

validation_records = []

for _, row in cash_flow_ratio_base.iterrows():

    company_id = int(row["CompanyID"])
    period_id = int(row["PeriodID"])

    operating_cash_flow = float(row["Operating Cash Flow"])
    capital_expenditures = float(row["Capital Expenditures"])
    free_cash_flow = float(row["Free Cash Flow"])
    revenue = float(row["Revenue"])
    net_income = float(row["Net Income"])
    dividends_paid = float(row["DividendsPaid"])

    expected_free_cash_flow = operating_cash_flow + capital_expenditures

    free_cash_flow_difference = round(
        free_cash_flow - expected_free_cash_flow,
        2
    )

    expected_ocf_to_sales = (
        operating_cash_flow / revenue
        if revenue != 0
        else 0
    )

    expected_dividend_payout = (
        dividends_paid / net_income
        if net_income != 0
        else 0
    )

    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,
        "Revenue": round(revenue, 2),
        "NetIncome": round(net_income, 2),
        "DividendsPaid": round(dividends_paid, 2),
        "ExpectedOperatingCashFlowToSales": round(expected_ocf_to_sales, 4),
        "ExpectedDividendPayoutRatio": round(expected_dividend_payout, 4),
        "Status": "PASSED" if free_cash_flow_difference == 0 else "FAILED"
    })

cash_flow_ratios_validation = pd.DataFrame(validation_records)

cash_flow_ratios_validation.to_csv(
    base_path / "Cash_Flow_Ratios_Validation_Summary.csv",
    index=False
)

print("Cash_Flow_Ratios_Validation_Summary.csv created successfully.")

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

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

print("Cash flow ratio rows:", len(fact_cash_flow_ratios))
print("Validation rows:", len(cash_flow_ratios_validation))

print()
print("Rows by RatioName:")
print(
    fact_cash_flow_ratios
    .groupby("RatioName")["RatioValue"]
    .count()
    .reset_index(name="NumberOfRows")
)

print()
print("Cash Flow Ratio Summary by Company:")
print(
    fact_cash_flow_ratios[
        fact_cash_flow_ratios["RatioName"].isin(
            [
                "Operating Cash Flow to Sales",
                "Free Cash Flow to Operating Cash Flow",
                "Free Cash Flow Margin",
                "Operating Cash Flow Margin",
                "Cash Flow Coverage Ratio",
                "Dividend Payout Ratio"
            ]
        )
    ]
    .groupby(["CompanyID", "CompanyName", "RatioName"])["RatioValue"]
    .mean()
    .round(4)
    .reset_index(name="AverageRatio")
)

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

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

print()
print("Preview of Fact_Cash_Flow_Ratios:")
print(fact_cash_flow_ratios.head(20))

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

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

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

Expected Output

The script or instructions create:

Resultado Esperado

El script o las instrucciones crean:

financial_ratios_bi_training/Fact_Cash_Flow_Ratios.csv

Validation Checklist

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 14: Cash Flow Indicator Ratios.

from pathlib import Path
import pandas as pd

# ============================================================
# Financial Ratios Analysis in BI
# Topic 14 - Cash Flow Indicator Ratios
# ============================================================

# ------------------------------------------------------------
# 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."
    )

cash_flow_path = base_path / "Fact_Cash_Flow.csv"
income_statement_path = base_path / "Fact_Income_Statement.csv"
balance_sheet_path = base_path / "Fact_Balance_Sheet.csv"
market_data_path = base_path / "Fact_Market_Data.csv"
dim_company_path = base_path / "Dim_Company.csv"
dim_period_path = base_path / "Dim_Period.csv"

required_files = [
    cash_flow_path,
    income_statement_path,
    balance_sheet_path,
    market_data_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
# ------------------------------------------------------------

cash_flow = pd.read_csv(cash_flow_path)
income_statement = pd.read_csv(income_statement_path)
balance_sheet = pd.read_csv(balance_sheet_path)
market_data = pd.read_csv(market_data_path)
dim_company = pd.read_csv(dim_company_path)
dim_period = pd.read_csv(dim_period_path)

print("Fact_Cash_Flow loaded:", len(cash_flow), "rows")
print("Fact_Income_Statement loaded:", len(income_statement), "rows")
print("Fact_Balance_Sheet loaded:", len(balance_sheet), "rows")
print("Fact_Market_Data loaded:", len(market_data), "rows")

# ------------------------------------------------------------
# Step 3 - Extract cash flow inputs
# ------------------------------------------------------------

cash_flow_inputs = cash_flow[
    cash_flow["RatioInput"].isin(
        [
            "Operating Cash Flow",
            "Capital Expenditures",
            "Free Cash Flow",
            "Dividends Paid",
            "Net Cash Flow After Dividends",
            "Cash Flow to Debt",
            "Cash Flow Coverage"
        ]
    )
].copy()

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

required_cash_flow_columns = [
    "Operating Cash Flow",
    "Capital Expenditures",
    "Free Cash Flow",
    "Dividends Paid",
    "Net Cash Flow After Dividends",
    "Cash Flow to Debt",
    "Cash Flow Coverage"
]

for column in required_cash_flow_columns:
    if column not in cash_flow_pivot.columns:
        cash_flow_pivot[column] = 0

# ------------------------------------------------------------
# Step 4 - Extract income statement inputs
# ------------------------------------------------------------

income_inputs = income_statement[
    income_statement["RatioInput"].isin(
        [
            "Revenue",
            "Net Income"
        ]
    )
].copy()

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

for column in ["Revenue", "Net Income"]:
    if column not in income_pivot.columns:
        income_pivot[column] = 0

# ------------------------------------------------------------
# Step 5 - Extract balance sheet inputs
# ------------------------------------------------------------

balance_inputs = balance_sheet[
    balance_sheet["RatioInput"].isin(
        [
            "Total Liabilities",
            "Total Assets",
            "Total Equity"
        ]
    )
].copy()

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

for column in ["Total Liabilities", "Total Assets", "Total Equity"]:
    if column not in balance_pivot.columns:
        balance_pivot[column] = 0

# ------------------------------------------------------------
# Step 6 - Extract market inputs
# ------------------------------------------------------------

market_inputs = market_data[
    [
        "CompanyID",
        "PeriodID",
        "DividendsPaid",
        "MarketCapitalization",
        "EnterpriseValue"
    ]
].copy()

# ------------------------------------------------------------
# Step 7 - Create base table
# ------------------------------------------------------------

cash_flow_ratio_base = (
    cash_flow_pivot
    .merge(income_pivot, on=["CompanyID", "PeriodID"], how="left")
    .merge(balance_pivot, on=["CompanyID", "PeriodID"], how="left")
    .merge(market_inputs, on=["CompanyID", "PeriodID"], how="left")
)

cash_flow_ratio_base = cash_flow_ratio_base.fillna(0)

# ------------------------------------------------------------
# Step 8 - Calculate cash flow ratios
# ------------------------------------------------------------

records = []

for _, row in cash_flow_ratio_base.iterrows():

    company_id = int(row["CompanyID"])
    period_id = int(row["PeriodID"])

    operating_cash_flow = float(row["Operating Cash Flow"])
    capital_expenditures = float(row["Capital Expenditures"])
    free_cash_flow = float(row["Free Cash Flow"])
    dividends_paid_cash_flow = abs(float(row["Dividends Paid"]))
    net_cash_flow_after_dividends = float(row["Net Cash Flow After Dividends"])

    revenue = float(row["Revenue"])
    net_income = float(row["Net Income"])

    total_liabilities = float(row["Total Liabilities"])
    total_assets = float(row["Total Assets"])
    total_equity = float(row["Total Equity"])

    dividends_paid_market = float(row["DividendsPaid"])
    market_cap = float(row["MarketCapitalization"])
    enterprise_value = float(row["EnterpriseValue"])

    operating_cash_flow_to_sales = (
        operating_cash_flow / revenue
        if revenue != 0
        else 0
    )

    free_cash_flow_to_operating_cash_flow = (
        free_cash_flow / operating_cash_flow
        if operating_cash_flow != 0
        else 0
    )

    free_cash_flow_margin = (
        free_cash_flow / revenue
        if revenue != 0
        else 0
    )

    operating_cash_flow_margin = (
        operating_cash_flow / revenue
        if revenue != 0
        else 0
    )

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

    dividend_payout_ratio = (
        dividends_paid_market / net_income
        if net_income != 0
        else 0
    )

    cash_return_on_assets = (
        operating_cash_flow / total_assets
        if total_assets != 0
        else 0
    )

    cash_return_on_equity = (
        operating_cash_flow / total_equity
        if total_equity != 0
        else 0
    )

    free_cash_flow_yield = (
        free_cash_flow / market_cap
        if market_cap != 0
        else 0
    )

    cash_flow_to_enterprise_value = (
        operating_cash_flow / enterprise_value
        if enterprise_value != 0
        else 0
    )

    records.extend([
        {
            "CompanyID": company_id,
            "PeriodID": period_id,
            "RatioCategory": "Cash Flow",
            "RatioName": "Operating Cash Flow",
            "RatioValue": round(operating_cash_flow, 2),
            "RatioFormat": "Currency",
            "Interpretation": "Cash generated from core operating activities."
        },
        {
            "CompanyID": company_id,
            "PeriodID": period_id,
            "RatioCategory": "Cash Flow",
            "RatioName": "Free Cash Flow",
            "RatioValue": round(free_cash_flow, 2),
            "RatioFormat": "Currency",
            "Interpretation": "Operating cash flow minus capital expenditures."
        },
        {
            "CompanyID": company_id,
            "PeriodID": period_id,
            "RatioCategory": "Cash Flow",
            "RatioName": "Operating Cash Flow to Sales",
            "RatioValue": round(operating_cash_flow_to_sales, 4),
            "RatioFormat": "Percent",
            "Interpretation": "Measures operating cash flow generated from each dollar of sales."
        },
        {
            "CompanyID": company_id,
            "PeriodID": period_id,
            "RatioCategory": "Cash Flow",
            "RatioName": "Free Cash Flow to Operating Cash Flow",
            "RatioValue": round(free_cash_flow_to_operating_cash_flow, 4),
            "RatioFormat": "Percent",
            "Interpretation": "Measures how much operating cash flow remains after capital expenditures."
        },
        {
            "CompanyID": company_id,
            "PeriodID": period_id,
            "RatioCategory": "Cash Flow",
            "RatioName": "Free Cash Flow Margin",
            "RatioValue": round(free_cash_flow_margin, 4),
            "RatioFormat": "Percent",
            "Interpretation": "Measures free cash flow generated as a percentage of revenue."
        },
        {
            "CompanyID": company_id,
            "PeriodID": period_id,
            "RatioCategory": "Cash Flow",
            "RatioName": "Operating Cash Flow Margin",
            "RatioValue": round(operating_cash_flow_margin, 4),
            "RatioFormat": "Percent",
            "Interpretation": "Measures operating cash flow as a percentage of revenue."
        },
        {
            "CompanyID": company_id,
            "PeriodID": period_id,
            "RatioCategory": "Cash Flow",
            "RatioName": "Cash Flow Coverage Ratio",
            "RatioValue": round(cash_flow_coverage_ratio, 4),
            "RatioFormat": "Decimal",
            "Interpretation": "Measures ability to cover total liabilities with operating cash flow."
        },
        {
            "CompanyID": company_id,
            "PeriodID": period_id,
            "RatioCategory": "Cash Flow",
            "RatioName": "Dividend Payout Ratio",
            "RatioValue": round(dividend_payout_ratio, 4),
            "RatioFormat": "Percent",
            "Interpretation": "Measures dividends paid as a percentage of net income."
        },
        {
            "CompanyID": company_id,
            "PeriodID": period_id,
            "RatioCategory": "Cash Flow",
            "RatioName": "Cash Return on Assets",
            "RatioValue": round(cash_return_on_assets, 4),
            "RatioFormat": "Percent",
            "Interpretation": "Measures operating cash flow generated by total assets."
        },
        {
            "CompanyID": company_id,
            "PeriodID": period_id,
            "RatioCategory": "Cash Flow",
            "RatioName": "Cash Return on Equity",
            "RatioValue": round(cash_return_on_equity, 4),
            "RatioFormat": "Percent",
            "Interpretation": "Measures operating cash flow generated by shareholder equity."
        },
        {
            "CompanyID": company_id,
            "PeriodID": period_id,
            "RatioCategory": "Cash Flow",
            "RatioName": "Free Cash Flow Yield",
            "RatioValue": round(free_cash_flow_yield, 4),
            "RatioFormat": "Percent",
            "Interpretation": "Measures free cash flow relative to market capitalization."
        },
        {
            "CompanyID": company_id,
            "PeriodID": period_id,
            "RatioCategory": "Cash Flow",
            "RatioName": "Cash Flow to Enterprise Value",
            "RatioValue": round(cash_flow_to_enterprise_value, 4),
            "RatioFormat": "Percent",
            "Interpretation": "Measures operating cash flow relative to enterprise value."
        }
    ])

fact_cash_flow_ratios = pd.DataFrame(records)

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

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

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

fact_cash_flow_ratios = fact_cash_flow_ratios[
    [
        "CompanyID",
        "CompanyName",
        "Industry",
        "PeriodID",
        "FiscalYear",
        "FiscalQuarter",
        "YearQuarter",
        "PeriodEndDate",
        "RatioCategory",
        "RatioName",
        "RatioValue",
        "RatioFormat",
        "Interpretation"
    ]
]

# ------------------------------------------------------------
# Step 10 - Export Fact_Cash_Flow_Ratios
# ------------------------------------------------------------

fact_cash_flow_ratios.to_csv(
    base_path / "Fact_Cash_Flow_Ratios.csv",
    index=False
)

print("Fact_Cash_Flow_Ratios.csv created successfully.")

# ------------------------------------------------------------
# Step 11 - Create validation summary
# ------------------------------------------------------------

validation_records = []

for _, row in cash_flow_ratio_base.iterrows():

    company_id = int(row["CompanyID"])
    period_id = int(row["PeriodID"])

    operating_cash_flow = float(row["Operating Cash Flow"])
    capital_expenditures = float(row["Capital Expenditures"])
    free_cash_flow = float(row["Free Cash Flow"])
    revenue = float(row["Revenue"])
    net_income = float(row["Net Income"])
    dividends_paid = float(row["DividendsPaid"])

    expected_free_cash_flow = operating_cash_flow + capital_expenditures

    free_cash_flow_difference = round(
        free_cash_flow - expected_free_cash_flow,
        2
    )

    expected_ocf_to_sales = (
        operating_cash_flow / revenue
        if revenue != 0
        else 0
    )

    expected_dividend_payout = (
        dividends_paid / net_income
        if net_income != 0
        else 0
    )

    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,
        "Revenue": round(revenue, 2),
        "NetIncome": round(net_income, 2),
        "DividendsPaid": round(dividends_paid, 2),
        "ExpectedOperatingCashFlowToSales": round(expected_ocf_to_sales, 4),
        "ExpectedDividendPayoutRatio": round(expected_dividend_payout, 4),
        "Status": "PASSED" if free_cash_flow_difference == 0 else "FAILED"
    })

cash_flow_ratios_validation = pd.DataFrame(validation_records)

cash_flow_ratios_validation.to_csv(
    base_path / "Cash_Flow_Ratios_Validation_Summary.csv",
    index=False
)

print("Cash_Flow_Ratios_Validation_Summary.csv created successfully.")

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

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

print("Cash flow ratio rows:", len(fact_cash_flow_ratios))
print("Validation rows:", len(cash_flow_ratios_validation))

print()
print("Rows by RatioName:")
print(
    fact_cash_flow_ratios
    .groupby("RatioName")["RatioValue"]
    .count()
    .reset_index(name="NumberOfRows")
)

print()
print("Cash Flow Ratio Summary by Company:")
print(
    fact_cash_flow_ratios[
        fact_cash_flow_ratios["RatioName"].isin(
            [
                "Operating Cash Flow to Sales",
                "Free Cash Flow to Operating Cash Flow",
                "Free Cash Flow Margin",
                "Operating Cash Flow Margin",
                "Cash Flow Coverage Ratio",
                "Dividend Payout Ratio"
            ]
        )
    ]
    .groupby(["CompanyID", "CompanyName", "RatioName"])["RatioValue"]
    .mean()
    .round(4)
    .reset_index(name="AverageRatio")
)

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

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

print()
print("Preview of Fact_Cash_Flow_Ratios:")
print(fact_cash_flow_ratios.head(20))

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

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

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