Objective
Create Fact_Income_Statement.csv using the synthetic trial balance and account mapping created in previous topics.
Production / Cybersecurity Warning: This exercise uses synthetic data only. Do not run scripts in production, real folders, corporate databases, or accounting environments without authorization, backups, testing, least-privilege permissions, change-control approval, and compliance with cybersecurity protocols.
Business Scenario
The trial balance contains accounting balances by account. Now BI needs an income statement that summarizes company performance by period.
Trial BalanceIncome Statement AccountsRevenue / ExpensesNet IncomeRatios
Step-by-Step Practice
Step 1 — Load trial balance, classify accounts, and calculate income statement lines
from pathlib import Path
import pandas as pd
# ============================================================
# Financial Ratios Analysis in BI
# Topic 6 - Build the Income Statement
# ============================================================
# ------------------------------------------------------------
# Step 1 - Set project folder
# ------------------------------------------------------------
base_path = Path("financial_ratios_bi_training")
if not base_path.exists():
raise FileNotFoundError(
"The project folder does not exist. Please run Topic 1 first."
)
trial_balance_path = base_path / "Fact_Trial_Balance.csv"
mapping_path = base_path / "Mapping_Financial_Statements.csv"
if not trial_balance_path.exists():
raise FileNotFoundError(
"Fact_Trial_Balance.csv was not found. Please run Topic 3 first."
)
if not mapping_path.exists():
raise FileNotFoundError(
"Mapping_Financial_Statements.csv was not found. Please run Topic 5 first."
)
print(f"Project folder found: {base_path.resolve()}")
# ------------------------------------------------------------
# Step 2 - Load source files
# ------------------------------------------------------------
trial_balance = pd.read_csv(trial_balance_path)
mapping = pd.read_csv(mapping_path)
trial_balance["AccountNumber"] = trial_balance["AccountNumber"].astype(str)
mapping["AccountNumber"] = mapping["AccountNumber"].astype(str)
print("Fact_Trial_Balance loaded:", len(trial_balance), "rows")
print("Mapping_Financial_Statements loaded:", len(mapping), "rows")
# ------------------------------------------------------------
# Step 3 - Filter income statement accounts
# ------------------------------------------------------------
income_mapping = mapping[
mapping["FinancialStatement"] == "Income Statement"
].copy()
income_trial_balance = trial_balance.merge(
income_mapping[
[
"AccountNumber",
"StatementSection",
"StatementLine",
"LineGroup",
"LineOrder",
"SignMultiplier",
"RatioInput"
]
],
on="AccountNumber",
how="inner"
)
print("Income statement source rows:", len(income_trial_balance))
# ------------------------------------------------------------
# Step 4 - Calculate signed amount
# ------------------------------------------------------------
# In this training model:
# - Revenue and Other Income are credit-normal accounts.
# - Expenses are debit-normal accounts.
#
# For reporting:
# - Revenue is shown as positive.
# - Expenses are shown as positive.
# - Calculated profit lines are derived separately.
income_trial_balance["Amount"] = income_trial_balance.apply(
lambda row: row["Credit"] if row["Credit"] > 0 else row["Debit"],
axis=1
)
income_trial_balance["Amount"] = income_trial_balance["Amount"].round(2)
# ------------------------------------------------------------
# Step 5 - Aggregate base income statement lines
# ------------------------------------------------------------
base_lines = (
income_trial_balance
.groupby(
[
"CompanyID",
"PeriodID",
"StatementSection",
"StatementLine",
"LineGroup",
"LineOrder",
"RatioInput"
]
)
.agg(Amount=("Amount", "sum"))
.reset_index()
)
base_lines["Amount"] = base_lines["Amount"].round(2)
base_lines["LineType"] = "Base Line"
# ------------------------------------------------------------
# Step 6 - Create calculated income statement lines
# ------------------------------------------------------------
calculated_records = []
for (company_id, period_id), group in base_lines.groupby(["CompanyID", "PeriodID"]):
revenue = group.loc[group["RatioInput"] == "Revenue", "Amount"].sum()
other_income = group.loc[group["RatioInput"] == "Other Income", "Amount"].sum()
cogs = group.loc[group["RatioInput"] == "COGS", "Amount"].sum()
operating_expenses = group.loc[group["RatioInput"] == "Operating Expenses", "Amount"].sum()
depreciation_expense = group.loc[group["RatioInput"] == "Depreciation Expense", "Amount"].sum()
interest_expense = group.loc[group["RatioInput"] == "Interest Expense", "Amount"].sum()
tax_expense = group.loc[group["RatioInput"] == "Tax Expense", "Amount"].sum()
total_operating_expenses = operating_expenses + depreciation_expense
gross_profit = revenue - cogs
operating_income = gross_profit - total_operating_expenses
pretax_income = operating_income + other_income - interest_expense
net_income = pretax_income - tax_expense
calculated_records.extend([
{
"CompanyID": company_id,
"PeriodID": period_id,
"StatementSection": "Profitability",
"StatementLine": "Gross Profit",
"LineGroup": "Calculated Profit Lines",
"LineOrder": 750,
"RatioInput": "Gross Profit",
"Amount": round(gross_profit, 2),
"LineType": "Calculated Line"
},
{
"CompanyID": company_id,
"PeriodID": period_id,
"StatementSection": "Profitability",
"StatementLine": "Total Operating Expenses",
"LineGroup": "Calculated Profit Lines",
"LineOrder": 850,
"RatioInput": "Total Operating Expenses",
"Amount": round(total_operating_expenses, 2),
"LineType": "Calculated Line"
},
{
"CompanyID": company_id,
"PeriodID": period_id,
"StatementSection": "Profitability",
"StatementLine": "Operating Income",
"LineGroup": "Calculated Profit Lines",
"LineOrder": 900,
"RatioInput": "Operating Income",
"Amount": round(operating_income, 2),
"LineType": "Calculated Line"
},
{
"CompanyID": company_id,
"PeriodID": period_id,
"StatementSection": "Profitability",
"StatementLine": "Pretax Income",
"LineGroup": "Calculated Profit Lines",
"LineOrder": 950,
"RatioInput": "Pretax Income",
"Amount": round(pretax_income, 2),
"LineType": "Calculated Line"
},
{
"CompanyID": company_id,
"PeriodID": period_id,
"StatementSection": "Profitability",
"StatementLine": "Net Income",
"LineGroup": "Calculated Profit Lines",
"LineOrder": 1100,
"RatioInput": "Net Income",
"Amount": round(net_income, 2),
"LineType": "Calculated Line"
}
])
calculated_lines = pd.DataFrame(calculated_records)
# ------------------------------------------------------------
# Step 7 - Combine base and calculated lines
# ------------------------------------------------------------
fact_income_statement = pd.concat(
[
base_lines,
calculated_lines
],
ignore_index=True
)
fact_income_statement = fact_income_statement.sort_values(
[
"CompanyID",
"PeriodID",
"LineOrder",
"StatementLine"
]
).reset_index(drop=True)
# ------------------------------------------------------------
# Step 8 - Add company and period labels
# ------------------------------------------------------------
dim_company_path = base_path / "Dim_Company.csv"
dim_period_path = base_path / "Dim_Period.csv"
dim_company = pd.read_csv(dim_company_path)
dim_period = pd.read_csv(dim_period_path)
fact_income_statement = fact_income_statement.merge(
dim_company[["CompanyID", "CompanyName", "Industry"]],
on="CompanyID",
how="left"
)
fact_income_statement = fact_income_statement.merge(
dim_period[["PeriodID", "FiscalYear", "FiscalQuarter", "YearQuarter", "PeriodEndDate"]],
on="PeriodID",
how="left"
)
fact_income_statement = fact_income_statement[
[
"CompanyID",
"CompanyName",
"Industry",
"PeriodID",
"FiscalYear",
"FiscalQuarter",
"YearQuarter",
"PeriodEndDate",
"StatementSection",
"StatementLine",
"LineGroup",
"LineOrder",
"RatioInput",
"LineType",
"Amount"
]
]
# ------------------------------------------------------------
# Step 9 - Export Fact_Income_Statement
# ------------------------------------------------------------
fact_income_statement.to_csv(
base_path / "Fact_Income_Statement.csv",
index=False
)
print("Fact_Income_Statement.csv created successfully.")
# ------------------------------------------------------------
# Step 10 - Create income statement validation summary
# ------------------------------------------------------------
income_validation = []
for (company_id, period_id), group in fact_income_statement.groupby(["CompanyID", "PeriodID"]):
revenue = group.loc[group["RatioInput"] == "Revenue", "Amount"].sum()
cogs = group.loc[group["RatioInput"] == "COGS", "Amount"].sum()
gross_profit = group.loc[group["RatioInput"] == "Gross Profit", "Amount"].sum()
operating_income = group.loc[group["RatioInput"] == "Operating Income", "Amount"].sum()
net_income = group.loc[group["RatioInput"] == "Net Income", "Amount"].sum()
gross_margin = round(gross_profit / revenue, 4) if revenue != 0 else 0
net_margin = round(net_income / revenue, 4) if revenue != 0 else 0
income_validation.append({
"CompanyID": company_id,
"PeriodID": period_id,
"Revenue": round(revenue, 2),
"COGS": round(cogs, 2),
"GrossProfit": round(gross_profit, 2),
"OperatingIncome": round(operating_income, 2),
"NetIncome": round(net_income, 2),
"GrossMargin": gross_margin,
"NetMargin": net_margin
})
income_validation = pd.DataFrame(income_validation)
income_validation.to_csv(
base_path / "Income_Statement_Validation_Summary.csv",
index=False
)
print("Income_Statement_Validation_Summary.csv created successfully.")
# ------------------------------------------------------------
# Step 11 - Print validation summary
# ------------------------------------------------------------
print()
print("==============================")
print("VALIDATION SUMMARY")
print("==============================")
print("Income statement rows:", len(fact_income_statement))
print("Validation rows:", len(income_validation))
print()
print("Rows by LineType:")
print(
fact_income_statement
.groupby("LineType")["StatementLine"]
.count()
.reset_index(name="NumberOfRows")
)
print()
print("Income Statement Lines:")
print(
fact_income_statement[
[
"LineOrder",
"StatementLine",
"LineType"
]
]
.drop_duplicates()
.sort_values("LineOrder")
)
print()
print("Sample validation summary:")
print(income_validation.head(10))
print()
print("Files currently in project folder:")
for file in base_path.glob("*.csv"):
print("-", file.name)
print()
print("Topic 6 completed successfully.")Expected Output
The script creates:
financial_ratios_bi_training/Fact_Income_Statement.csvValidation
The output should contain one row per company and period, with core lines such as Revenue, COGS, Gross Profit, Operating Income, Income Before Tax, Tax Expense, and Net Income.
Business Interpretation
This topic converts accounting detail into business performance. Net income becomes the foundation for profitability ratios such as profit margin, return on assets, and return on equity.
Final Result
After Topic 6, the model has its first financial statement fact table. Topic 7 will build the balance sheet.