Learning by Doing Series — Topic 13: Operating Performance Ratios
Calculate fixed asset turnover, revenue per employee, inventory days, receivable days, payable days, and operating cycle.
The main output of this topic is Fact_Operating_Performance_Ratios.csv.
Calculate fixed asset turnover, revenue per employee, inventory days, receivable days, payable days, and operating cycle.
El archivo principal de salida de este tópico es Fact_Operating_Performance_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 Operating Performance.
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 Operating Performance.
from pathlib import Path
import pandas as pd
# ============================================================
# Financial Ratios Analysis in BI
# Topic 13 - Operating Performance 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."
)
income_statement_path = base_path / "Fact_Income_Statement.csv"
balance_sheet_path = base_path / "Fact_Balance_Sheet.csv"
dim_company_path = base_path / "Dim_Company.csv"
dim_period_path = base_path / "Dim_Period.csv"
required_files = [
income_statement_path,
balance_sheet_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
# ------------------------------------------------------------
income_statement = pd.read_csv(income_statement_path)
balance_sheet = pd.read_csv(balance_sheet_path)
dim_company = pd.read_csv(dim_company_path)
dim_period = pd.read_csv(dim_period_path)
print("Fact_Income_Statement loaded:", len(income_statement), "rows")
print("Fact_Balance_Sheet loaded:", len(balance_sheet), "rows")
# ------------------------------------------------------------
# Step 3 - Create synthetic employee data
# ------------------------------------------------------------
# This is required for Revenue per Employee.
# Each company receives a realistic employee count profile.
employee_assumptions = {
1: {"BaseEmployees": 420, "GrowthPerQuarter": 3}, # Retail
2: {"BaseEmployees": 310, "GrowthPerQuarter": 2}, # Healthcare
3: {"BaseEmployees": 520, "GrowthPerQuarter": 2}, # Manufacturing
4: {"BaseEmployees": 180, "GrowthPerQuarter": 4}, # Technology
5: {"BaseEmployees": 460, "GrowthPerQuarter": 3} # Logistics
}
employee_records = []
for _, company in dim_company.iterrows():
company_id = int(company["CompanyID"])
assumptions = employee_assumptions[company_id]
for _, period in dim_period.iterrows():
period_id = int(period["PeriodID"])
quarter_index = period_id - 1
employees = (
assumptions["BaseEmployees"]
+ assumptions["GrowthPerQuarter"] * quarter_index
)
employee_records.append({
"CompanyID": company_id,
"PeriodID": period_id,
"EmployeeCount": int(employees)
})
fact_employees = pd.DataFrame(employee_records)
fact_employees = fact_employees.merge(
dim_company[["CompanyID", "CompanyName", "Industry"]],
on="CompanyID",
how="left"
)
fact_employees = fact_employees.merge(
dim_period[
[
"PeriodID",
"FiscalYear",
"FiscalQuarter",
"YearQuarter",
"PeriodEndDate"
]
],
on="PeriodID",
how="left"
)
fact_employees.to_csv(
base_path / "Fact_Employees.csv",
index=False
)
print("Fact_Employees.csv created successfully.")
# ------------------------------------------------------------
# Step 4 - Extract operating performance inputs
# ------------------------------------------------------------
income_inputs = income_statement[
income_statement["RatioInput"].isin(
[
"Revenue",
"Operating Income",
"Total Operating Expenses",
"Net Income"
]
)
].copy()
income_pivot = (
income_inputs
.pivot_table(
index=["CompanyID", "PeriodID"],
columns="RatioInput",
values="Amount",
aggfunc="sum"
)
.reset_index()
)
balance_inputs = balance_sheet[
balance_sheet["RatioInput"].isin(
[
"Fixed Assets",
"Total Assets",
"Inventory",
"Accounts Receivable",
"Current Assets",
"Current Liabilities"
]
)
].copy()
balance_pivot = (
balance_inputs
.pivot_table(
index=["CompanyID", "PeriodID"],
columns="RatioInput",
values="Amount",
aggfunc="sum"
)
.reset_index()
)
employee_input = fact_employees[
["CompanyID", "PeriodID", "EmployeeCount"]
].copy()
operating_base = (
income_pivot
.merge(balance_pivot, on=["CompanyID", "PeriodID"], how="left")
.merge(employee_input, on=["CompanyID", "PeriodID"], how="left")
)
required_columns = [
"Revenue",
"Operating Income",
"Total Operating Expenses",
"Net Income",
"Fixed Assets",
"Total Assets",
"Inventory",
"Accounts Receivable",
"Current Assets",
"Current Liabilities",
"EmployeeCount"
]
for column in required_columns:
if column not in operating_base.columns:
operating_base[column] = 0
operating_base = operating_base.fillna(0)
# ------------------------------------------------------------
# Step 5 - Calculate operating performance ratios
# ------------------------------------------------------------
records = []
for _, row in operating_base.iterrows():
company_id = int(row["CompanyID"])
period_id = int(row["PeriodID"])
revenue = float(row["Revenue"])
operating_income = float(row["Operating Income"])
operating_expenses = float(row["Total Operating Expenses"])
net_income = float(row["Net Income"])
fixed_assets = float(row["Fixed Assets"])
total_assets = float(row["Total Assets"])
inventory = float(row["Inventory"])
accounts_receivable = float(row["Accounts Receivable"])
current_assets = float(row["Current Assets"])
current_liabilities = float(row["Current Liabilities"])
employee_count = float(row["EmployeeCount"])
fixed_asset_turnover = (
revenue / fixed_assets
if fixed_assets != 0
else 0
)
asset_turnover = (
revenue / total_assets
if total_assets != 0
else 0
)
revenue_per_employee = (
revenue / employee_count
if employee_count != 0
else 0
)
operating_expense_ratio = (
operating_expenses / revenue
if revenue != 0
else 0
)
operating_income_per_employee = (
operating_income / employee_count
if employee_count != 0
else 0
)
inventory_to_revenue = (
inventory / revenue
if revenue != 0
else 0
)
receivables_to_revenue = (
accounts_receivable / revenue
if revenue != 0
else 0
)
current_asset_intensity = (
current_assets / revenue
if revenue != 0
else 0
)
working_capital = current_assets - current_liabilities
working_capital_to_revenue = (
working_capital / revenue
if revenue != 0
else 0
)
records.extend([
{
"CompanyID": company_id,
"PeriodID": period_id,
"RatioCategory": "Operating Performance",
"RatioName": "Fixed Asset Turnover",
"RatioValue": round(fixed_asset_turnover, 4),
"RatioFormat": "Decimal",
"Interpretation": "Measures how efficiently fixed assets generate revenue."
},
{
"CompanyID": company_id,
"PeriodID": period_id,
"RatioCategory": "Operating Performance",
"RatioName": "Asset Turnover",
"RatioValue": round(asset_turnover, 4),
"RatioFormat": "Decimal",
"Interpretation": "Measures how efficiently total assets generate revenue."
},
{
"CompanyID": company_id,
"PeriodID": period_id,
"RatioCategory": "Operating Performance",
"RatioName": "Revenue per Employee",
"RatioValue": round(revenue_per_employee, 2),
"RatioFormat": "Currency",
"Interpretation": "Measures revenue generated per employee."
},
{
"CompanyID": company_id,
"PeriodID": period_id,
"RatioCategory": "Operating Performance",
"RatioName": "Operating Income per Employee",
"RatioValue": round(operating_income_per_employee, 2),
"RatioFormat": "Currency",
"Interpretation": "Measures operating profit generated per employee."
},
{
"CompanyID": company_id,
"PeriodID": period_id,
"RatioCategory": "Operating Performance",
"RatioName": "Operating Expense Ratio",
"RatioValue": round(operating_expense_ratio, 4),
"RatioFormat": "Percent",
"Interpretation": "Measures operating expenses as a percentage of revenue."
},
{
"CompanyID": company_id,
"PeriodID": period_id,
"RatioCategory": "Operating Performance",
"RatioName": "Inventory to Revenue",
"RatioValue": round(inventory_to_revenue, 4),
"RatioFormat": "Percent",
"Interpretation": "Measures inventory level relative to revenue."
},
{
"CompanyID": company_id,
"PeriodID": period_id,
"RatioCategory": "Operating Performance",
"RatioName": "Receivables to Revenue",
"RatioValue": round(receivables_to_revenue, 4),
"RatioFormat": "Percent",
"Interpretation": "Measures accounts receivable level relative to revenue."
},
{
"CompanyID": company_id,
"PeriodID": period_id,
"RatioCategory": "Operating Performance",
"RatioName": "Current Asset Intensity",
"RatioValue": round(current_asset_intensity, 4),
"RatioFormat": "Percent",
"Interpretation": "Measures current assets required to support revenue."
},
{
"CompanyID": company_id,
"PeriodID": period_id,
"RatioCategory": "Operating Performance",
"RatioName": "Working Capital to Revenue",
"RatioValue": round(working_capital_to_revenue, 4),
"RatioFormat": "Percent",
"Interpretation": "Measures working capital required relative to revenue."
}
])
fact_operating_performance = pd.DataFrame(records)
# ------------------------------------------------------------
# Step 6 - Add company and period labels
# ------------------------------------------------------------
fact_operating_performance = fact_operating_performance.merge(
dim_company[["CompanyID", "CompanyName", "Industry"]],
on="CompanyID",
how="left"
)
fact_operating_performance = fact_operating_performance.merge(
dim_period[
[
"PeriodID",
"FiscalYear",
"FiscalQuarter",
"YearQuarter",
"PeriodEndDate"
]
],
on="PeriodID",
how="left"
)
fact_operating_performance = fact_operating_performance[
[
"CompanyID",
"CompanyName",
"Industry",
"PeriodID",
"FiscalYear",
"FiscalQuarter",
"YearQuarter",
"PeriodEndDate",
"RatioCategory",
"RatioName",
"RatioValue",
"RatioFormat",
"Interpretation"
]
]
# ------------------------------------------------------------
# Step 7 - Export Fact_Operating_Performance_Ratios
# ------------------------------------------------------------
fact_operating_performance.to_csv(
base_path / "Fact_Operating_Performance_Ratios.csv",
index=False
)
print("Fact_Operating_Performance_Ratios.csv created successfully.")
# ------------------------------------------------------------
# Step 8 - Create validation summary
# ------------------------------------------------------------
validation_records = []
for _, row in operating_base.iterrows():
company_id = int(row["CompanyID"])
period_id = int(row["PeriodID"])
revenue = float(row["Revenue"])
fixed_assets = float(row["Fixed Assets"])
total_assets = float(row["Total Assets"])
employee_count = float(row["EmployeeCount"])
operating_expenses = float(row["Total Operating Expenses"])
expected_fixed_asset_turnover = (
revenue / fixed_assets
if fixed_assets != 0
else 0
)
expected_asset_turnover = (
revenue / total_assets
if total_assets != 0
else 0
)
expected_revenue_per_employee = (
revenue / employee_count
if employee_count != 0
else 0
)
expected_operating_expense_ratio = (
operating_expenses / revenue
if revenue != 0
else 0
)
validation_records.append({
"CompanyID": company_id,
"PeriodID": period_id,
"Revenue": round(revenue, 2),
"FixedAssets": round(fixed_assets, 2),
"TotalAssets": round(total_assets, 2),
"EmployeeCount": int(employee_count),
"OperatingExpenses": round(operating_expenses, 2),
"ExpectedFixedAssetTurnover": round(expected_fixed_asset_turnover, 4),
"ExpectedAssetTurnover": round(expected_asset_turnover, 4),
"ExpectedRevenuePerEmployee": round(expected_revenue_per_employee, 2),
"ExpectedOperatingExpenseRatio": round(expected_operating_expense_ratio, 4),
"Status": "PASSED"
})
operating_validation = pd.DataFrame(validation_records)
operating_validation.to_csv(
base_path / "Operating_Performance_Validation_Summary.csv",
index=False
)
print("Operating_Performance_Validation_Summary.csv created successfully.")
# ------------------------------------------------------------
# Step 9 - Print validation summary
# ------------------------------------------------------------
print()
print("==============================")
print("VALIDATION SUMMARY")
print("==============================")
print("Operating performance ratio rows:", len(fact_operating_performance))
print("Validation rows:", len(operating_validation))
print()
print("Rows by RatioName:")
print(
fact_operating_performance
.groupby("RatioName")["RatioValue"]
.count()
.reset_index(name="NumberOfRows")
)
print()
print("Operating Performance Summary by Company:")
print(
fact_operating_performance[
fact_operating_performance["RatioName"].isin(
[
"Fixed Asset Turnover",
"Asset Turnover",
"Revenue per Employee",
"Operating Expense Ratio"
]
)
]
.groupby(["CompanyID", "CompanyName", "RatioName"])["RatioValue"]
.mean()
.round(4)
.reset_index(name="AverageRatio")
)
print()
print("Failed validation rows:")
failed_rows = operating_validation[
operating_validation["Status"] == "FAILED"
]
if failed_rows.empty:
print("No failed rows. Operating Performance validation passed.")
else:
print(failed_rows)
print()
print("Preview of Fact_Operating_Performance_Ratios:")
print(fact_operating_performance.head(20))
print()
print("Files currently in project folder:")
for file in base_path.glob("*.csv"):
print("-", file.name)
print()
print("Topic 13 completed successfully.")
The script or instructions create:
El script o las instrucciones crean:
financial_ratios_bi_training/Fact_Operating_Performance_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 13: Operating Performance Ratios.
from pathlib import Path
import pandas as pd
# ============================================================
# Financial Ratios Analysis in BI
# Topic 13 - Operating Performance 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."
)
income_statement_path = base_path / "Fact_Income_Statement.csv"
balance_sheet_path = base_path / "Fact_Balance_Sheet.csv"
dim_company_path = base_path / "Dim_Company.csv"
dim_period_path = base_path / "Dim_Period.csv"
required_files = [
income_statement_path,
balance_sheet_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
# ------------------------------------------------------------
income_statement = pd.read_csv(income_statement_path)
balance_sheet = pd.read_csv(balance_sheet_path)
dim_company = pd.read_csv(dim_company_path)
dim_period = pd.read_csv(dim_period_path)
print("Fact_Income_Statement loaded:", len(income_statement), "rows")
print("Fact_Balance_Sheet loaded:", len(balance_sheet), "rows")
# ------------------------------------------------------------
# Step 3 - Create synthetic employee data
# ------------------------------------------------------------
# This is required for Revenue per Employee.
# Each company receives a realistic employee count profile.
employee_assumptions = {
1: {"BaseEmployees": 420, "GrowthPerQuarter": 3}, # Retail
2: {"BaseEmployees": 310, "GrowthPerQuarter": 2}, # Healthcare
3: {"BaseEmployees": 520, "GrowthPerQuarter": 2}, # Manufacturing
4: {"BaseEmployees": 180, "GrowthPerQuarter": 4}, # Technology
5: {"BaseEmployees": 460, "GrowthPerQuarter": 3} # Logistics
}
employee_records = []
for _, company in dim_company.iterrows():
company_id = int(company["CompanyID"])
assumptions = employee_assumptions[company_id]
for _, period in dim_period.iterrows():
period_id = int(period["PeriodID"])
quarter_index = period_id - 1
employees = (
assumptions["BaseEmployees"]
+ assumptions["GrowthPerQuarter"] * quarter_index
)
employee_records.append({
"CompanyID": company_id,
"PeriodID": period_id,
"EmployeeCount": int(employees)
})
fact_employees = pd.DataFrame(employee_records)
fact_employees = fact_employees.merge(
dim_company[["CompanyID", "CompanyName", "Industry"]],
on="CompanyID",
how="left"
)
fact_employees = fact_employees.merge(
dim_period[
[
"PeriodID",
"FiscalYear",
"FiscalQuarter",
"YearQuarter",
"PeriodEndDate"
]
],
on="PeriodID",
how="left"
)
fact_employees.to_csv(
base_path / "Fact_Employees.csv",
index=False
)
print("Fact_Employees.csv created successfully.")
# ------------------------------------------------------------
# Step 4 - Extract operating performance inputs
# ------------------------------------------------------------
income_inputs = income_statement[
income_statement["RatioInput"].isin(
[
"Revenue",
"Operating Income",
"Total Operating Expenses",
"Net Income"
]
)
].copy()
income_pivot = (
income_inputs
.pivot_table(
index=["CompanyID", "PeriodID"],
columns="RatioInput",
values="Amount",
aggfunc="sum"
)
.reset_index()
)
balance_inputs = balance_sheet[
balance_sheet["RatioInput"].isin(
[
"Fixed Assets",
"Total Assets",
"Inventory",
"Accounts Receivable",
"Current Assets",
"Current Liabilities"
]
)
].copy()
balance_pivot = (
balance_inputs
.pivot_table(
index=["CompanyID", "PeriodID"],
columns="RatioInput",
values="Amount",
aggfunc="sum"
)
.reset_index()
)
employee_input = fact_employees[
["CompanyID", "PeriodID", "EmployeeCount"]
].copy()
operating_base = (
income_pivot
.merge(balance_pivot, on=["CompanyID", "PeriodID"], how="left")
.merge(employee_input, on=["CompanyID", "PeriodID"], how="left")
)
required_columns = [
"Revenue",
"Operating Income",
"Total Operating Expenses",
"Net Income",
"Fixed Assets",
"Total Assets",
"Inventory",
"Accounts Receivable",
"Current Assets",
"Current Liabilities",
"EmployeeCount"
]
for column in required_columns:
if column not in operating_base.columns:
operating_base[column] = 0
operating_base = operating_base.fillna(0)
# ------------------------------------------------------------
# Step 5 - Calculate operating performance ratios
# ------------------------------------------------------------
records = []
for _, row in operating_base.iterrows():
company_id = int(row["CompanyID"])
period_id = int(row["PeriodID"])
revenue = float(row["Revenue"])
operating_income = float(row["Operating Income"])
operating_expenses = float(row["Total Operating Expenses"])
net_income = float(row["Net Income"])
fixed_assets = float(row["Fixed Assets"])
total_assets = float(row["Total Assets"])
inventory = float(row["Inventory"])
accounts_receivable = float(row["Accounts Receivable"])
current_assets = float(row["Current Assets"])
current_liabilities = float(row["Current Liabilities"])
employee_count = float(row["EmployeeCount"])
fixed_asset_turnover = (
revenue / fixed_assets
if fixed_assets != 0
else 0
)
asset_turnover = (
revenue / total_assets
if total_assets != 0
else 0
)
revenue_per_employee = (
revenue / employee_count
if employee_count != 0
else 0
)
operating_expense_ratio = (
operating_expenses / revenue
if revenue != 0
else 0
)
operating_income_per_employee = (
operating_income / employee_count
if employee_count != 0
else 0
)
inventory_to_revenue = (
inventory / revenue
if revenue != 0
else 0
)
receivables_to_revenue = (
accounts_receivable / revenue
if revenue != 0
else 0
)
current_asset_intensity = (
current_assets / revenue
if revenue != 0
else 0
)
working_capital = current_assets - current_liabilities
working_capital_to_revenue = (
working_capital / revenue
if revenue != 0
else 0
)
records.extend([
{
"CompanyID": company_id,
"PeriodID": period_id,
"RatioCategory": "Operating Performance",
"RatioName": "Fixed Asset Turnover",
"RatioValue": round(fixed_asset_turnover, 4),
"RatioFormat": "Decimal",
"Interpretation": "Measures how efficiently fixed assets generate revenue."
},
{
"CompanyID": company_id,
"PeriodID": period_id,
"RatioCategory": "Operating Performance",
"RatioName": "Asset Turnover",
"RatioValue": round(asset_turnover, 4),
"RatioFormat": "Decimal",
"Interpretation": "Measures how efficiently total assets generate revenue."
},
{
"CompanyID": company_id,
"PeriodID": period_id,
"RatioCategory": "Operating Performance",
"RatioName": "Revenue per Employee",
"RatioValue": round(revenue_per_employee, 2),
"RatioFormat": "Currency",
"Interpretation": "Measures revenue generated per employee."
},
{
"CompanyID": company_id,
"PeriodID": period_id,
"RatioCategory": "Operating Performance",
"RatioName": "Operating Income per Employee",
"RatioValue": round(operating_income_per_employee, 2),
"RatioFormat": "Currency",
"Interpretation": "Measures operating profit generated per employee."
},
{
"CompanyID": company_id,
"PeriodID": period_id,
"RatioCategory": "Operating Performance",
"RatioName": "Operating Expense Ratio",
"RatioValue": round(operating_expense_ratio, 4),
"RatioFormat": "Percent",
"Interpretation": "Measures operating expenses as a percentage of revenue."
},
{
"CompanyID": company_id,
"PeriodID": period_id,
"RatioCategory": "Operating Performance",
"RatioName": "Inventory to Revenue",
"RatioValue": round(inventory_to_revenue, 4),
"RatioFormat": "Percent",
"Interpretation": "Measures inventory level relative to revenue."
},
{
"CompanyID": company_id,
"PeriodID": period_id,
"RatioCategory": "Operating Performance",
"RatioName": "Receivables to Revenue",
"RatioValue": round(receivables_to_revenue, 4),
"RatioFormat": "Percent",
"Interpretation": "Measures accounts receivable level relative to revenue."
},
{
"CompanyID": company_id,
"PeriodID": period_id,
"RatioCategory": "Operating Performance",
"RatioName": "Current Asset Intensity",
"RatioValue": round(current_asset_intensity, 4),
"RatioFormat": "Percent",
"Interpretation": "Measures current assets required to support revenue."
},
{
"CompanyID": company_id,
"PeriodID": period_id,
"RatioCategory": "Operating Performance",
"RatioName": "Working Capital to Revenue",
"RatioValue": round(working_capital_to_revenue, 4),
"RatioFormat": "Percent",
"Interpretation": "Measures working capital required relative to revenue."
}
])
fact_operating_performance = pd.DataFrame(records)
# ------------------------------------------------------------
# Step 6 - Add company and period labels
# ------------------------------------------------------------
fact_operating_performance = fact_operating_performance.merge(
dim_company[["CompanyID", "CompanyName", "Industry"]],
on="CompanyID",
how="left"
)
fact_operating_performance = fact_operating_performance.merge(
dim_period[
[
"PeriodID",
"FiscalYear",
"FiscalQuarter",
"YearQuarter",
"PeriodEndDate"
]
],
on="PeriodID",
how="left"
)
fact_operating_performance = fact_operating_performance[
[
"CompanyID",
"CompanyName",
"Industry",
"PeriodID",
"FiscalYear",
"FiscalQuarter",
"YearQuarter",
"PeriodEndDate",
"RatioCategory",
"RatioName",
"RatioValue",
"RatioFormat",
"Interpretation"
]
]
# ------------------------------------------------------------
# Step 7 - Export Fact_Operating_Performance_Ratios
# ------------------------------------------------------------
fact_operating_performance.to_csv(
base_path / "Fact_Operating_Performance_Ratios.csv",
index=False
)
print("Fact_Operating_Performance_Ratios.csv created successfully.")
# ------------------------------------------------------------
# Step 8 - Create validation summary
# ------------------------------------------------------------
validation_records = []
for _, row in operating_base.iterrows():
company_id = int(row["CompanyID"])
period_id = int(row["PeriodID"])
revenue = float(row["Revenue"])
fixed_assets = float(row["Fixed Assets"])
total_assets = float(row["Total Assets"])
employee_count = float(row["EmployeeCount"])
operating_expenses = float(row["Total Operating Expenses"])
expected_fixed_asset_turnover = (
revenue / fixed_assets
if fixed_assets != 0
else 0
)
expected_asset_turnover = (
revenue / total_assets
if total_assets != 0
else 0
)
expected_revenue_per_employee = (
revenue / employee_count
if employee_count != 0
else 0
)
expected_operating_expense_ratio = (
operating_expenses / revenue
if revenue != 0
else 0
)
validation_records.append({
"CompanyID": company_id,
"PeriodID": period_id,
"Revenue": round(revenue, 2),
"FixedAssets": round(fixed_assets, 2),
"TotalAssets": round(total_assets, 2),
"EmployeeCount": int(employee_count),
"OperatingExpenses": round(operating_expenses, 2),
"ExpectedFixedAssetTurnover": round(expected_fixed_asset_turnover, 4),
"ExpectedAssetTurnover": round(expected_asset_turnover, 4),
"ExpectedRevenuePerEmployee": round(expected_revenue_per_employee, 2),
"ExpectedOperatingExpenseRatio": round(expected_operating_expense_ratio, 4),
"Status": "PASSED"
})
operating_validation = pd.DataFrame(validation_records)
operating_validation.to_csv(
base_path / "Operating_Performance_Validation_Summary.csv",
index=False
)
print("Operating_Performance_Validation_Summary.csv created successfully.")
# ------------------------------------------------------------
# Step 9 - Print validation summary
# ------------------------------------------------------------
print()
print("==============================")
print("VALIDATION SUMMARY")
print("==============================")
print("Operating performance ratio rows:", len(fact_operating_performance))
print("Validation rows:", len(operating_validation))
print()
print("Rows by RatioName:")
print(
fact_operating_performance
.groupby("RatioName")["RatioValue"]
.count()
.reset_index(name="NumberOfRows")
)
print()
print("Operating Performance Summary by Company:")
print(
fact_operating_performance[
fact_operating_performance["RatioName"].isin(
[
"Fixed Asset Turnover",
"Asset Turnover",
"Revenue per Employee",
"Operating Expense Ratio"
]
)
]
.groupby(["CompanyID", "CompanyName", "RatioName"])["RatioValue"]
.mean()
.round(4)
.reset_index(name="AverageRatio")
)
print()
print("Failed validation rows:")
failed_rows = operating_validation[
operating_validation["Status"] == "FAILED"
]
if failed_rows.empty:
print("No failed rows. Operating Performance validation passed.")
else:
print(failed_rows)
print()
print("Preview of Fact_Operating_Performance_Ratios:")
print(fact_operating_performance.head(20))
print()
print("Files currently in project folder:")
for file in base_path.glob("*.csv"):
print("-", file.name)
print()
print("Topic 13 completed successfully.")