Financial Ratios Analysis in BI

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

Copyfrom pathlib import Path
import pandas as pd

# ============================================================
# Financial Ratios Analysis in BI
# Topic 19 - Executive Financial Dashboard Dataset
# ============================================================

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

required_files = [
    "Fact_Income_Statement.csv",
    "Fact_Balance_Sheet.csv",
    "Fact_Cash_Flow.csv",
    "Fact_Market_Data.csv",
    "Fact_Liquidity_Ratios.csv",
    "Fact_Profitability_Ratios.csv",
    "Fact_Debt_Ratios.csv",
    "Fact_Operating_Performance_Ratios.csv",
    "Fact_Cash_Flow_Ratios.csv",
    "Fact_Investment_Valuation_Ratios.csv",
    "Dim_Company.csv",
    "Dim_Period.csv"
]

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

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

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

income_statement = pd.read_csv(base_path / "Fact_Income_Statement.csv")
balance_sheet = pd.read_csv(base_path / "Fact_Balance_Sheet.csv")
cash_flow = pd.read_csv(base_path / "Fact_Cash_Flow.csv")
market_data = pd.read_csv(base_path / "Fact_Market_Data.csv")

liquidity_ratios = pd.read_csv(base_path / "Fact_Liquidity_Ratios.csv")
profitability_ratios = pd.read_csv(base_path / "Fact_Profitability_Ratios.csv")
debt_ratios = pd.read_csv(base_path / "Fact_Debt_Ratios.csv")
operating_ratios = pd.read_csv(base_path / "Fact_Operating_Performance_Ratios.csv")
cash_flow_ratios = pd.read_csv(base_path / "Fact_Cash_Flow_Ratios.csv")
valuation_ratios = pd.read_csv(base_path / "Fact_Investment_Valuation_Ratios.csv")

dim_company = pd.read_csv(base_path / "Dim_Company.csv")
dim_period = pd.read_csv(base_path / "Dim_Period.csv")

print("Source files loaded successfully.")

# ------------------------------------------------------------
# Step 3 - Helper function to extract RatioInput values
# ------------------------------------------------------------

def pivot_financial_lines(df, ratio_inputs, value_column="Amount"):
    filtered = df[df["RatioInput"].isin(ratio_inputs)].copy()

    pivot = (
        filtered
        .pivot_table(
            index=["CompanyID", "PeriodID"],
            columns="RatioInput",
            values=value_column,
            aggfunc="sum"
        )
        .reset_index()
    )

    for ratio_input in ratio_inputs:
        if ratio_input not in pivot.columns:
            pivot[ratio_input] = 0

    return pivot


def pivot_ratio_values(df, ratio_names):
    filtered = df[df["RatioName"].isin(ratio_names)].copy()

    pivot = (
        filtered
        .pivot_table(
            index=["CompanyID", "PeriodID"],
            columns="RatioName",
            values="RatioValue",
            aggfunc="mean"
        )
        .reset_index()
    )

    for ratio_name in ratio_names:
        if ratio_name not in pivot.columns:
            pivot[ratio_name] = 0

    return pivot


# ------------------------------------------------------------
# Step 4 - Extract executive financial statement metrics
# ------------------------------------------------------------

income_pivot = pivot_financial_lines(
    income_statement,
    [
        "Revenue",
        "Gross Profit",
        "Operating Income",
        "Pretax Income",
        "Net Income",
        "Total Operating Expenses"
    ]
)

balance_pivot = pivot_financial_lines(
    balance_sheet,
    [
        "Current Assets",
        "Current Liabilities",
        "Working Capital",
        "Total Assets",
        "Total Liabilities",
        "Total Equity",
        "Total Liabilities and Equity"
    ]
)

cash_flow_pivot = pivot_financial_lines(
    cash_flow,
    [
        "Operating Cash Flow",
        "Capital Expenditures",
        "Free Cash Flow",
        "Net Cash Flow After Dividends",
        "Cash Flow to Debt",
        "Cash Flow Coverage"
    ]
)

# ------------------------------------------------------------
# Step 5 - Extract executive ratio metrics
# ------------------------------------------------------------

liquidity_pivot = pivot_ratio_values(
    liquidity_ratios,
    [
        "Current Ratio",
        "Quick Ratio",
        "Cash Ratio",
        "Working Capital"
    ]
)

profitability_pivot = pivot_ratio_values(
    profitability_ratios,
    [
        "Gross Margin",
        "Operating Margin",
        "Net Profit Margin",
        "Effective Tax Rate",
        "Return on Assets",
        "Return on Equity",
        "Return on Capital Employed"
    ]
)

debt_pivot = pivot_ratio_values(
    debt_ratios,
    [
        "Debt Ratio",
        "Debt-to-Equity Ratio",
        "Liabilities to Equity Ratio",
        "Capitalization Ratio",
        "Interest Coverage Ratio",
        "Cash Flow to Debt Ratio"
    ]
)

operating_pivot = pivot_ratio_values(
    operating_ratios,
    [
        "Fixed Asset Turnover",
        "Asset Turnover",
        "Revenue per Employee",
        "Operating Income per Employee",
        "Operating Expense Ratio",
        "Inventory to Revenue",
        "Receivables to Revenue"
    ]
)

cash_flow_ratio_pivot = pivot_ratio_values(
    cash_flow_ratios,
    [
        "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",
        "Cash Return on Assets",
        "Cash Return on Equity",
        "Free Cash Flow Yield"
    ]
)

valuation_pivot = pivot_ratio_values(
    valuation_ratios,
    [
        "Price / Earnings Ratio",
        "Price / Book Ratio",
        "Price / Sales Ratio",
        "Dividend Yield",
        "Earnings Yield",
        "Enterprise Value / Sales",
        "Enterprise Value / Operating Income",
        "Market Cap / Equity",
        "Market Cap / Assets"
    ]
)

# ------------------------------------------------------------
# Step 6 - Prepare market data
# ------------------------------------------------------------

market_columns = [
    "CompanyID",
    "PeriodID",
    "SharePrice",
    "SharesOutstanding",
    "MarketCapitalization",
    "EnterpriseValue",
    "EPS",
    "BookValuePerShare",
    "DividendPerShare",
    "DividendYield",
    "PriceToEarnings",
    "PriceToBook"
]

market_pivot = market_data[market_columns].copy()

# ------------------------------------------------------------
# Step 7 - Build executive dashboard base
# ------------------------------------------------------------

dashboard = (
    income_pivot
    .merge(balance_pivot, on=["CompanyID", "PeriodID"], how="left")
    .merge(cash_flow_pivot, on=["CompanyID", "PeriodID"], how="left")
    .merge(market_pivot, on=["CompanyID", "PeriodID"], how="left")
    .merge(liquidity_pivot, on=["CompanyID", "PeriodID"], how="left")
    .merge(profitability_pivot, on=["CompanyID", "PeriodID"], how="left")
    .merge(debt_pivot, on=["CompanyID", "PeriodID"], how="left")
    .merge(operating_pivot, on=["CompanyID", "PeriodID"], how="left")
    .merge(cash_flow_ratio_pivot, on=["CompanyID", "PeriodID"], how="left")
    .merge(valuation_pivot, on=["CompanyID", "PeriodID"], how="left")
)

dashboard = dashboard.fillna(0)

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

dashboard = dashboard.merge(
    dim_company[
        [
            "CompanyID",
            "CompanyName",
            "Ticker",
            "Industry",
            "BusinessProfile",
            "RiskProfile",
            "Country"
        ]
    ],
    on="CompanyID",
    how="left"
)

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

# ------------------------------------------------------------
# Step 9 - Calculate executive health scores
# ------------------------------------------------------------

def score_current_ratio(value):
    if value >= 1.8:
        return 20
    if value >= 1.2:
        return 15
    if value >= 1.0:
        return 10
    return 5


def score_profitability(value):
    if value >= 0.15:
        return 20
    if value >= 0.08:
        return 15
    if value >= 0.03:
        return 10
    return 5


def score_debt_ratio(value):
    if value <= 0.35:
        return 20
    if value <= 0.55:
        return 15
    if value <= 0.70:
        return 10
    return 5


def score_cash_flow(value):
    if value >= 0.15:
        return 20
    if value >= 0.08:
        return 15
    if value >= 0.03:
        return 10
    return 5


def score_valuation(value):
    if value <= 15 and value > 0:
        return 20
    if value <= 25 and value > 0:
        return 15
    if value <= 40 and value > 0:
        return 10
    return 5


dashboard["LiquidityScore"] = dashboard["Current Ratio"].apply(score_current_ratio)
dashboard["ProfitabilityScore"] = dashboard["Net Profit Margin"].apply(score_profitability)
dashboard["DebtScore"] = dashboard["Debt Ratio"].apply(score_debt_ratio)
dashboard["CashFlowScore"] = dashboard["Operating Cash Flow to Sales"].apply(score_cash_flow)
dashboard["ValuationScore"] = dashboard["Price / Earnings Ratio"].apply(score_valuation)

dashboard["FinancialHealthScore"] = (
    dashboard["LiquidityScore"]
    + dashboard["ProfitabilityScore"]
    + dashboard["DebtScore"]
    + dashboard["CashFlowScore"]
    + dashboard["ValuationScore"]
)

def health_category(score):
    if score >= 85:
        return "Strong"
    if score >= 70:
        return "Healthy"
    if score >= 55:
        return "Watch"
    return "Risk"

dashboard["FinancialHealthCategory"] = dashboard["FinancialHealthScore"].apply(health_category)

# ------------------------------------------------------------
# Step 10 - Add executive flags
# ------------------------------------------------------------

dashboard["LiquidityFlag"] = dashboard["Current Ratio"].apply(
    lambda x: "Good" if x >= 1.2 else "Review"
)

dashboard["ProfitabilityFlag"] = dashboard["Net Profit Margin"].apply(
    lambda x: "Good" if x >= 0.05 else "Review"
)

dashboard["DebtFlag"] = dashboard["Debt Ratio"].apply(
    lambda x: "Good" if x <= 0.60 else "Review"
)

dashboard["CashFlowFlag"] = dashboard["Operating Cash Flow"].apply(
    lambda x: "Good" if x > 0 else "Review"
)

dashboard["ValuationFlag"] = dashboard["Price / Earnings Ratio"].apply(
    lambda x: "Good" if 0 < x <= 35 else "Review"
)

dashboard["ExecutiveAlertCount"] = (
    (dashboard["LiquidityFlag"] == "Review").astype(int)
    + (dashboard["ProfitabilityFlag"] == "Review").astype(int)
    + (dashboard["DebtFlag"] == "Review").astype(int)
    + (dashboard["CashFlowFlag"] == "Review").astype(int)
    + (dashboard["ValuationFlag"] == "Review").astype(int)
)

dashboard["ExecutiveStatus"] = dashboard["ExecutiveAlertCount"].apply(
    lambda x: "No Major Alerts" if x == 0 else f"{x} Area(s) Need Review"
)

# ------------------------------------------------------------
# Step 11 - Reorder dashboard columns
# ------------------------------------------------------------

preferred_columns = [
    "CompanyID",
    "CompanyName",
    "Ticker",
    "Industry",
    "BusinessProfile",
    "RiskProfile",
    "Country",
    "PeriodID",
    "FiscalYear",
    "FiscalQuarter",
    "YearQuarter",
    "PeriodEndDate",

    "Revenue",
    "Gross Profit",
    "Operating Income",
    "Net Income",
    "Total Operating Expenses",

    "Current Assets",
    "Current Liabilities",
    "Working Capital",
    "Total Assets",
    "Total Liabilities",
    "Total Equity",

    "Operating Cash Flow",
    "Capital Expenditures",
    "Free Cash Flow",
    "Net Cash Flow After Dividends",

    "SharePrice",
    "SharesOutstanding",
    "MarketCapitalization",
    "EnterpriseValue",
    "EPS",
    "BookValuePerShare",
    "DividendPerShare",

    "Current Ratio",
    "Quick Ratio",
    "Cash Ratio",
    "Gross Margin",
    "Operating Margin",
    "Net Profit Margin",
    "Return on Assets",
    "Return on Equity",
    "Debt Ratio",
    "Debt-to-Equity Ratio",
    "Interest Coverage Ratio",
    "Operating Cash Flow to Sales",
    "Free Cash Flow Margin",
    "Dividend Payout Ratio",
    "Fixed Asset Turnover",
    "Asset Turnover",
    "Revenue per Employee",
    "Price / Earnings Ratio",
    "Price / Book Ratio",
    "Price / Sales Ratio",
    "Dividend Yield",
    "Enterprise Value / Sales",

    "LiquidityScore",
    "ProfitabilityScore",
    "DebtScore",
    "CashFlowScore",
    "ValuationScore",
    "FinancialHealthScore",
    "FinancialHealthCategory",
    "LiquidityFlag",
    "ProfitabilityFlag",
    "DebtFlag",
    "CashFlowFlag",
    "ValuationFlag",
    "ExecutiveAlertCount",
    "ExecutiveStatus"
]

existing_columns = [col for col in preferred_columns if col in dashboard.columns]
remaining_columns = [col for col in dashboard.columns if col not in existing_columns]

dashboard = dashboard[existing_columns + remaining_columns]

# ------------------------------------------------------------
# Step 12 - Export Executive_Financial_Dashboard
# ------------------------------------------------------------

dashboard.to_csv(
    base_path / "Executive_Financial_Dashboard.csv",
    index=False
)

print("Executive_Financial_Dashboard.csv created successfully.")

# ------------------------------------------------------------
# Step 13 - Create Executive KPI Summary
# ------------------------------------------------------------

latest_period_id = dashboard["PeriodID"].max()

latest_dashboard = dashboard[
    dashboard["PeriodID"] == latest_period_id
].copy()

executive_kpi_summary = (
    latest_dashboard
    .groupby(["CompanyID", "CompanyName", "Ticker", "Industry"])
    .agg(
        LatestPeriod=("YearQuarter", "max"),
        Revenue=("Revenue", "sum"),
        NetIncome=("Net Income", "sum"),
        TotalAssets=("Total Assets", "sum"),
        TotalLiabilities=("Total Liabilities", "sum"),
        TotalEquity=("Total Equity", "sum"),
        OperatingCashFlow=("Operating Cash Flow", "sum"),
        FreeCashFlow=("Free Cash Flow", "sum"),
        MarketCapitalization=("MarketCapitalization", "sum"),
        EnterpriseValue=("EnterpriseValue", "sum"),
        CurrentRatio=("Current Ratio", "mean"),
        NetProfitMargin=("Net Profit Margin", "mean"),
        DebtRatio=("Debt Ratio", "mean"),
        ReturnOnAssets=("Return on Assets", "mean"),
        ReturnOnEquity=("Return on Equity", "mean"),
        PriceToEarningsRatio=("Price / Earnings Ratio", "mean"),
        DividendYield=("Dividend Yield", "mean"),
        FinancialHealthScore=("FinancialHealthScore", "mean"),
        ExecutiveAlertCount=("ExecutiveAlertCount", "sum")
    )
    .reset_index()
)

executive_kpi_summary["FinancialHealthScore"] = (
    executive_kpi_summary["FinancialHealthScore"].round(0).astype(int)
)

executive_kpi_summary["FinancialHealthCategory"] = (
    executive_kpi_summary["FinancialHealthScore"].apply(health_category)
)

executive_kpi_summary["ExecutiveStatus"] = executive_kpi_summary["ExecutiveAlertCount"].apply(
    lambda x: "No Major Alerts" if x == 0 else f"{x} Area(s) Need Review"
)

executive_kpi_summary.to_csv(
    base_path / "Executive_KPI_Summary.csv",
    index=False
)

print("Executive_KPI_Summary.csv created successfully.")

# ------------------------------------------------------------
# Step 14 - Create Company Financial Health Score trend
# ------------------------------------------------------------

company_health_score = dashboard[
    [
        "CompanyID",
        "CompanyName",
        "Ticker",
        "Industry",
        "PeriodID",
        "FiscalYear",
        "FiscalQuarter",
        "YearQuarter",
        "FinancialHealthScore",
        "FinancialHealthCategory",
        "LiquidityScore",
        "ProfitabilityScore",
        "DebtScore",
        "CashFlowScore",
        "ValuationScore",
        "ExecutiveAlertCount",
        "ExecutiveStatus"
    ]
].copy()

company_health_score.to_csv(
    base_path / "Company_Financial_Health_Score.csv",
    index=False
)

print("Company_Financial_Health_Score.csv created successfully.")

# ------------------------------------------------------------
# Step 15 - Create dashboard validation report
# ------------------------------------------------------------

validation_records = []

expected_rows = len(dim_company) * len(dim_period)
actual_rows = len(dashboard)

validation_records.append({
    "ValidationCheck": "Dashboard Row Count",
    "ExpectedValue": expected_rows,
    "ActualValue": actual_rows,
    "Difference": actual_rows - expected_rows,
    "Status": "PASSED" if actual_rows == expected_rows else "FAILED"
})

required_dashboard_columns = [
    "Revenue",
    "Net Income",
    "Total Assets",
    "Total Liabilities",
    "Total Equity",
    "Operating Cash Flow",
    "Free Cash Flow",
    "Current Ratio",
    "Net Profit Margin",
    "Debt Ratio",
    "Return on Assets",
    "Return on Equity",
    "Price / Earnings Ratio",
    "Dividend Yield",
    "FinancialHealthScore"
]

for column in required_dashboard_columns:
    missing_count = dashboard[column].isna().sum() if column in dashboard.columns else actual_rows

    validation_records.append({
        "ValidationCheck": f"Missing Values - {column}",
        "ExpectedValue": 0,
        "ActualValue": int(missing_count),
        "Difference": int(missing_count),
        "Status": "PASSED" if missing_count == 0 else "FAILED"
    })

dashboard_validation = pd.DataFrame(validation_records)

dashboard_validation.to_csv(
    base_path / "Executive_Dashboard_Validation.csv",
    index=False
)

print("Executive_Dashboard_Validation.csv created successfully.")

# ------------------------------------------------------------
# Step 16 - Print validation summary
# ------------------------------------------------------------

print()
print("==============================")
print("EXECUTIVE DASHBOARD SUMMARY")
print("==============================")

print("Dashboard rows:", len(dashboard))
print("Dashboard columns:", len(dashboard.columns))
print("Expected rows:", expected_rows)

print()
print("Financial Health by Latest Period:")
print(
    executive_kpi_summary[
        [
            "CompanyID",
            "CompanyName",
            "Ticker",
            "FinancialHealthScore",
            "FinancialHealthCategory",
            "ExecutiveStatus"
        ]
    ]
)

print()
print("Average Health Score by Industry:")
print(
    dashboard
    .groupby("Industry")["FinancialHealthScore"]
    .mean()
    .round(2)
    .reset_index(name="AverageFinancialHealthScore")
)

print()
print("Validation Summary:")
print(dashboard_validation)

print()
print("Files created:")

created_files = [
    "Executive_Financial_Dashboard.csv",
    "Executive_KPI_Summary.csv",
    "Company_Financial_Health_Score.csv",
    "Executive_Dashboard_Validation.csv"
]

for file_name in created_files:
    if (base_path / file_name).exists():
        print("-", file_name)

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

financial_ratios_bi_training/DAX_Financial_Measures.txt

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.

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.