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 16 - Data Quality Issues
# ============================================================

# ------------------------------------------------------------
# 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_Trial_Balance.csv",
    "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"
]

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
# ------------------------------------------------------------

trial_balance = pd.read_csv(base_path / "Fact_Trial_Balance.csv")
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")

print("Source files loaded successfully.")

# ------------------------------------------------------------
# Step 3 - Combine ratio files for global quality checks
# ------------------------------------------------------------

all_ratios = pd.concat(
    [
        liquidity_ratios,
        profitability_ratios,
        debt_ratios,
        operating_ratios,
        cash_flow_ratios,
        valuation_ratios
    ],
    ignore_index=True
)

print("Combined ratio rows:", len(all_ratios))

# ------------------------------------------------------------
# Step 4 - Create controlled data quality issues
# ------------------------------------------------------------
# This creates a copy of selected datasets with intentional issues.
# The original clean files are not modified.

trial_balance_issues = trial_balance.copy()
market_data_issues = market_data.copy()
ratio_issues = all_ratios.copy()

# Issue 1 - Missing value in Trial Balance AccountName
if len(trial_balance_issues) > 10:
    trial_balance_issues.loc[10, "AccountName"] = None

# Issue 2 - Duplicate trial balance row
duplicate_trial_balance_row = trial_balance_issues.iloc[[0]].copy()
trial_balance_issues = pd.concat(
    [trial_balance_issues, duplicate_trial_balance_row],
    ignore_index=True
)

# Issue 3 - Negative debit value
if len(trial_balance_issues) > 20:
    trial_balance_issues.loc[20, "Debit"] = -5000

# Issue 4 - Missing market share price
if len(market_data_issues) > 5:
    market_data_issues.loc[5, "SharePrice"] = None

# Issue 5 - Suspicious market capitalization
if len(market_data_issues) > 8:
    market_data_issues.loc[8, "MarketCapitalization"] = (
        market_data_issues.loc[8, "MarketCapitalization"] * 25
    )

# Issue 6 - Invalid ratio value
if len(ratio_issues) > 15:
    ratio_issues.loc[15, "RatioValue"] = -999

# Issue 7 - Missing ratio category
if len(ratio_issues) > 25:
    ratio_issues.loc[25, "RatioCategory"] = None

# Issue 8 - Extreme outlier ratio value
if len(ratio_issues) > 35:
    ratio_issues.loc[35, "RatioValue"] = 999999

# ------------------------------------------------------------
# Step 5 - Export issue datasets
# ------------------------------------------------------------

trial_balance_issues.to_csv(
    base_path / "DQ_Test_Fact_Trial_Balance_With_Issues.csv",
    index=False
)

market_data_issues.to_csv(
    base_path / "DQ_Test_Fact_Market_Data_With_Issues.csv",
    index=False
)

ratio_issues.to_csv(
    base_path / "DQ_Test_All_Ratios_With_Issues.csv",
    index=False
)

print("DQ test files created successfully.")

# ------------------------------------------------------------
# Step 6 - Detect issues
# ------------------------------------------------------------

issue_records = []
detail_records = []

# ------------------------------------------------------------
# Helper function
# ------------------------------------------------------------

def add_issue(issue_type, dataset_name, column_name, issue_count, severity, recommendation):
    issue_records.append({
        "IssueType": issue_type,
        "DatasetName": dataset_name,
        "ColumnName": column_name,
        "IssueCount": int(issue_count),
        "Severity": severity,
        "Recommendation": recommendation
    })

def add_detail(issue_type, dataset_name, row_identifier, column_name, issue_value, severity):
    detail_records.append({
        "IssueType": issue_type,
        "DatasetName": dataset_name,
        "RowIdentifier": row_identifier,
        "ColumnName": column_name,
        "IssueValue": issue_value,
        "Severity": severity
    })

# ------------------------------------------------------------
# Step 7 - Missing values checks
# ------------------------------------------------------------

datasets_to_check = {
    "DQ_Test_Fact_Trial_Balance_With_Issues": trial_balance_issues,
    "DQ_Test_Fact_Market_Data_With_Issues": market_data_issues,
    "DQ_Test_All_Ratios_With_Issues": ratio_issues
}

for dataset_name, df in datasets_to_check.items():

    missing_counts = df.isna().sum()

    for column_name, missing_count in missing_counts.items():

        if missing_count > 0:
            add_issue(
                issue_type="Missing Values",
                dataset_name=dataset_name,
                column_name=column_name,
                issue_count=missing_count,
                severity="High",
                recommendation="Review missing values before using this dataset in BI reports."
            )

            missing_rows = df[df[column_name].isna()]

            for idx, row in missing_rows.iterrows():
                add_detail(
                    issue_type="Missing Values",
                    dataset_name=dataset_name,
                    row_identifier=idx,
                    column_name=column_name,
                    issue_value="NULL",
                    severity="High"
                )

# ------------------------------------------------------------
# Step 8 - Duplicate trial balance rows
# ------------------------------------------------------------

duplicate_mask = trial_balance_issues.duplicated(
    subset=["CompanyID", "PeriodID", "AccountNumber"],
    keep=False
)

duplicate_rows = trial_balance_issues[duplicate_mask]

if len(duplicate_rows) > 0:
    add_issue(
        issue_type="Duplicate Company-Period-Account",
        dataset_name="DQ_Test_Fact_Trial_Balance_With_Issues",
        column_name="CompanyID / PeriodID / AccountNumber",
        issue_count=len(duplicate_rows),
        severity="High",
        recommendation="Each company-period-account combination should appear only once."
    )

    for idx, row in duplicate_rows.iterrows():
        add_detail(
            issue_type="Duplicate Company-Period-Account",
            dataset_name="DQ_Test_Fact_Trial_Balance_With_Issues",
            row_identifier=idx,
            column_name="CompanyID / PeriodID / AccountNumber",
            issue_value=f"{row['CompanyID']} / {row['PeriodID']} / {row['AccountNumber']}",
            severity="High"
        )

# ------------------------------------------------------------
# Step 9 - Negative debit / credit values
# ------------------------------------------------------------

negative_debits = trial_balance_issues[trial_balance_issues["Debit"] < 0]
negative_credits = trial_balance_issues[trial_balance_issues["Credit"] < 0]

if len(negative_debits) > 0:
    add_issue(
        issue_type="Negative Debit",
        dataset_name="DQ_Test_Fact_Trial_Balance_With_Issues",
        column_name="Debit",
        issue_count=len(negative_debits),
        severity="High",
        recommendation="Debit values should not be negative in this training trial balance format."
    )

    for idx, row in negative_debits.iterrows():
        add_detail(
            issue_type="Negative Debit",
            dataset_name="DQ_Test_Fact_Trial_Balance_With_Issues",
            row_identifier=idx,
            column_name="Debit",
            issue_value=row["Debit"],
            severity="High"
        )

if len(negative_credits) > 0:
    add_issue(
        issue_type="Negative Credit",
        dataset_name="DQ_Test_Fact_Trial_Balance_With_Issues",
        column_name="Credit",
        issue_count=len(negative_credits),
        severity="High",
        recommendation="Credit values should not be negative in this training trial balance format."
    )

# ------------------------------------------------------------
# Step 10 - Trial balance difference check
# ------------------------------------------------------------

tb_check = (
    trial_balance_issues
    .groupby(["CompanyID", "PeriodID"])
    .agg(
        TotalDebit=("Debit", "sum"),
        TotalCredit=("Credit", "sum")
    )
    .reset_index()
)

tb_check["Difference"] = (
    tb_check["TotalDebit"] - tb_check["TotalCredit"]
).round(2)

failed_tb = tb_check[tb_check["Difference"] != 0]

if len(failed_tb) > 0:
    add_issue(
        issue_type="Unbalanced Trial Balance",
        dataset_name="DQ_Test_Fact_Trial_Balance_With_Issues",
        column_name="Debit / Credit",
        issue_count=len(failed_tb),
        severity="Critical",
        recommendation="Investigate company-period combinations where debits do not equal credits."
    )

    for idx, row in failed_tb.iterrows():
        add_detail(
            issue_type="Unbalanced Trial Balance",
            dataset_name="DQ_Test_Fact_Trial_Balance_With_Issues",
            row_identifier=f"CompanyID {row['CompanyID']} / PeriodID {row['PeriodID']}",
            column_name="Difference",
            issue_value=row["Difference"],
            severity="Critical"
        )

# ------------------------------------------------------------
# Step 11 - Market data validation
# ------------------------------------------------------------

market_data_issues["ExpectedMarketCapitalization"] = (
    market_data_issues["SharePrice"]
    * market_data_issues["SharesOutstanding"]
).round(2)

market_data_issues["MarketCapDifference"] = (
    market_data_issues["MarketCapitalization"]
    - market_data_issues["ExpectedMarketCapitalization"]
).round(2)

bad_market_cap = market_data_issues[
    market_data_issues["MarketCapDifference"].abs() > 1
]

if len(bad_market_cap) > 0:
    add_issue(
        issue_type="Market Capitalization Mismatch",
        dataset_name="DQ_Test_Fact_Market_Data_With_Issues",
        column_name="MarketCapitalization",
        issue_count=len(bad_market_cap),
        severity="High",
        recommendation="Validate market capitalization against share price multiplied by shares outstanding."
    )

    for idx, row in bad_market_cap.iterrows():
        add_detail(
            issue_type="Market Capitalization Mismatch",
            dataset_name="DQ_Test_Fact_Market_Data_With_Issues",
            row_identifier=idx,
            column_name="MarketCapDifference",
            issue_value=row["MarketCapDifference"],
            severity="High"
        )

# ------------------------------------------------------------
# Step 12 - Invalid and suspicious ratio values
# ------------------------------------------------------------

missing_ratio_category = ratio_issues[ratio_issues["RatioCategory"].isna()]

if len(missing_ratio_category) > 0:
    add_issue(
        issue_type="Missing Ratio Category",
        dataset_name="DQ_Test_All_Ratios_With_Issues",
        column_name="RatioCategory",
        issue_count=len(missing_ratio_category),
        severity="Medium",
        recommendation="Every ratio should have a ratio category for proper BI grouping."
    )

negative_ratio_values = ratio_issues[ratio_issues["RatioValue"] < 0]

if len(negative_ratio_values) > 0:
    add_issue(
        issue_type="Negative Ratio Value",
        dataset_name="DQ_Test_All_Ratios_With_Issues",
        column_name="RatioValue",
        issue_count=len(negative_ratio_values),
        severity="Medium",
        recommendation="Review negative ratio values and confirm whether they are business-valid or data errors."
    )

    for idx, row in negative_ratio_values.iterrows():
        add_detail(
            issue_type="Negative Ratio Value",
            dataset_name="DQ_Test_All_Ratios_With_Issues",
            row_identifier=idx,
            column_name="RatioValue",
            issue_value=row["RatioValue"],
            severity="Medium"
        )

extreme_ratio_values = ratio_issues[ratio_issues["RatioValue"].abs() > 100000]

if len(extreme_ratio_values) > 0:
    add_issue(
        issue_type="Extreme Ratio Outlier",
        dataset_name="DQ_Test_All_Ratios_With_Issues",
        column_name="RatioValue",
        issue_count=len(extreme_ratio_values),
        severity="High",
        recommendation="Investigate extreme ratio values before using them in dashboards."
    )

    for idx, row in extreme_ratio_values.iterrows():
        add_detail(
            issue_type="Extreme Ratio Outlier",
            dataset_name="DQ_Test_All_Ratios_With_Issues",
            row_identifier=idx,
            column_name="RatioValue",
            issue_value=row["RatioValue"],
            severity="High"
        )

# ------------------------------------------------------------
# Step 13 - Create issue reports
# ------------------------------------------------------------

if issue_records:
    data_quality_issues = pd.DataFrame(issue_records)
else:
    data_quality_issues = pd.DataFrame([
        {
            "IssueType": "No Issues Found",
            "DatasetName": "All",
            "ColumnName": "All",
            "IssueCount": 0,
            "Severity": "None",
            "Recommendation": "No data quality issues detected."
        }
    ])

if detail_records:
    data_quality_details = pd.DataFrame(detail_records)
else:
    data_quality_details = pd.DataFrame([
        {
            "IssueType": "No Issues Found",
            "DatasetName": "All",
            "RowIdentifier": "N/A",
            "ColumnName": "All",
            "IssueValue": "N/A",
            "Severity": "None"
        }
    ])

data_quality_issues.to_csv(
    base_path / "Data_Quality_Issues.csv",
    index=False
)

data_quality_details.to_csv(
    base_path / "Data_Quality_Issues_Detail.csv",
    index=False
)

print("Data_Quality_Issues.csv created successfully.")
print("Data_Quality_Issues_Detail.csv created successfully.")

# ------------------------------------------------------------
# Step 14 - Print validation summary
# ------------------------------------------------------------

print()
print("==============================")
print("DATA QUALITY SUMMARY")
print("==============================")

print("Total issue categories:", len(data_quality_issues))
print("Total issue details:", len(data_quality_details))

print()
print("Issues by Severity:")
print(
    data_quality_issues
    .groupby("Severity")["IssueCount"]
    .sum()
    .reset_index(name="TotalIssues")
)

print()
print("Issues by Type:")
print(
    data_quality_issues[
        [
            "IssueType",
            "DatasetName",
            "ColumnName",
            "IssueCount",
            "Severity"
        ]
    ]
)

print()
print("Sample Issue Details:")
print(data_quality_details.head(20))

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

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

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

financial_ratios_bi_training/Data_Quality_Issues.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.

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.