Learning by Doing Series — Topic 14: Cash Flow Indicator Ratios
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.
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.
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.
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.
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.
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.
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.")
The script or instructions create:
El script o las instrucciones crean:
financial_ratios_bi_training/Fact_Cash_Flow_Ratios.csv
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.
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.
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.
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.")