Mini-project 3: Balance Sheet¶

#importations on top for code quality and organization
from pathlib import Path

import numpy as np
import pandas as pd
from IPython.display import display

pd.set_option('display.max_columns', 200)
pd.set_option('display.width', 180)
pd.set_option('display.float_format', lambda x: f'{x:,.2f}')

#Locating the base direction to get CSVs
def locate_base_dir(start: Path) -> Path:
    required_files = [
        Path('Mini-projects/outputs/unit1_tidy_historical_pack_usdm.csv'),
        Path('Mini-projects/outputs/unit1_tidy_historical_audit.csv'),
        Path('Mini projects/outputs/mini2_income_statement_full_model.csv'),
    ]
    for candidate in [start, *start.parents]:
        if all((candidate / rel_path).exists() for rel_path in required_files):
            return candidate
    raise FileNotFoundError('Could not locate the Chapter 4 tidy-data inputs from the current working directory.')


BASE_DIR = locate_base_dir(Path.cwd())
UNIT1_PACK_PATH = BASE_DIR / 'Mini-projects' / 'outputs' / 'unit1_tidy_historical_pack_usdm.csv'
UNIT1_AUDIT_PATH = BASE_DIR / 'Mini-projects' / 'outputs' / 'unit1_tidy_historical_audit.csv'
MINI2_PATH = BASE_DIR / 'Mini projects' / 'outputs' / 'mini2_income_statement_full_model.csv'
OUTPUT_DIR = BASE_DIR / 'Mini-projects' / 'outputs'
OUTPUT_DIR.mkdir(parents=True, exist_ok=True)

print(f'Base directory: {BASE_DIR}')
print(f'Unit 1 tidy pack: {UNIT1_PACK_PATH}')
print(f'Unit 1 audit: {UNIT1_AUDIT_PATH}')
print(f'Mini Project 2 model: {MINI2_PATH}')
print(f'Chapter 4 outputs will be written to: {OUTPUT_DIR}')

#using pd to read the csv
pack = pd.read_csv(UNIT1_PACK_PATH)
audit = pd.read_csv(UNIT1_AUDIT_PATH)
mini2 = pd.read_csv(MINI2_PATH)

#Setting the required columns we are going to use
required_pack_cols = {
    'Quarter', 'FiscalYear', 'QuarterNumber', 'Date', 'Cash', 'CashAndSTInvestments',
    'AccountsReceivable', 'Inventory', 'AccountsPayable', 'AccruedExpenses',
    'DeferredRevenue', 'OtherCurrentAssets', 'OtherCurrentLiabilities',
    'TotalCurrentAssets', 'PPENet', 'TotalAssets', 'TotalLiabilities', 'TotalEquity',
    'TotalDebt', 'CapEx', 'Dividends', 'SBC', 'DeltaWC_CFSign'
}
required_audit_cols = {'Quarter', 'AuditPass', 'BalanceError_USDm', 'GrossProfitError_USDm'}
required_mini2_cols = {
    'Period', 'FiscalYear', 'Quarter', 'Stage', 'Revenue', 'COGS', 'DA',
    'InterestExpense', 'NetIncome', 'EBITDA', 'EBIT', 'GrossProfit',
    'OperatingIncome', 'EBT', 'TaxExpense'
}

#As mentioned in first class we should keep guardrails of the code and notify if any of the other projects data broke and we are missing something
missing_pack = sorted(required_pack_cols - set(pack.columns))
missing_audit = sorted(required_audit_cols - set(audit.columns))
missing_mini2 = sorted(required_mini2_cols - set(mini2.columns))
if missing_pack:
    raise KeyError(f'Missing pack columns: {missing_pack}')
if missing_audit:
    raise KeyError(f'Missing audit columns: {missing_audit}')
if missing_mini2:
    raise KeyError(f'Missing mini2 columns: {missing_mini2}')

pack = pack.copy()
pack['ReportDate'] = pd.to_datetime(pack['Date'])
pack['PeriodKey'] = pack['FiscalYear'].astype(int).astype(str) + '-Q' + pack['QuarterNumber'].astype(int).astype(str)

mini2_hist = mini2.loc[mini2['Stage'].eq('Historical')].copy()
mini2_hist['PeriodKey'] = mini2_hist['FiscalYear'].astype(int).astype(str) + '-Q' + mini2_hist['Quarter'].astype(int).astype(str)

spine = pack.merge(
    mini2_hist[
        [
            'PeriodKey', 'Stage', 'Revenue', 'COGS', 'GrossProfit', 'OperatingIncome',
            'EBIT', 'DA', 'EBITDA', 'InterestExpense', 'EBT', 'TaxExpense', 'NetIncome'
        ]
    ],
    on='PeriodKey',
    how='inner',
    suffixes=('_pack', '_mini2'),
)
spine = spine.sort_values(['FiscalYear', 'QuarterNumber']).reset_index(drop=True)

if spine['PeriodKey'].duplicated().any():
    duplicates = sorted(spine.loc[spine['PeriodKey'].duplicated(keep=False), 'PeriodKey'].unique().tolist())
    raise ValueError(f'Duplicate merged periods found: {duplicates}')

#Asserting all the 40 quarters are loaded
assert len(pack) == 40, f'Expected 40 historical pack rows, found {len(pack)}'
assert len(mini2_hist) == 40, f'Expected 40 historical mini2 rows, found {len(mini2_hist)}'
assert len(spine) == 40, f'Expected 40 merged historical rows, found {len(spine)}'

source_summary = pd.DataFrame(
    [
        ('unit1_tidy_historical_pack_usdm.csv', len(pack), pack['Quarter'].min(), pack['Quarter'].max()),
        ('unit1_tidy_historical_audit.csv', len(audit), audit['Quarter'].min(), audit['Quarter'].max()),
        ('mini2_income_statement_full_model.csv (Historical only)', len(mini2_hist), mini2_hist['PeriodKey'].min(), mini2_hist['PeriodKey'].max()),
        ('Merged Chapter 4 modeling spine', len(spine), spine['Quarter'].min(), spine['Quarter'].max()),
    ],
    columns=['Source', 'Rows', 'FirstPeriod', 'LastPeriod'],
)


audit_exceptions = audit.loc[~audit['AuditPass']].copy()

display(source_summary)

print(f"Inherited Unit 1 audit passes: {int(audit['AuditPass'].sum())} of {len(audit)} quarters")
if audit_exceptions.empty:
    print('No inherited Unit 1 audit exceptions were found.')
else:
    print('Inherited Unit 1 audit exceptions are shown below. They come from the prior project output and do not stop the Chapter 4 build.')
    display(audit_exceptions)

#creating the df
balance_sheet = pd.DataFrame({
    'Quarter': spine['Quarter'],
    'PeriodKey': spine['PeriodKey'],
    'ReportDate': spine['ReportDate'].dt.strftime('%Y-%m-%d'),
    'ReportedCash': spine['Cash'],
    'ShortTermInvestments': working_capital['ShortTermInvestments'],
    'AR': working_capital['AR'],
    'Inventory': working_capital['Inventory'],
    'OtherCurrentAssets': working_capital['OtherCurrentAssets'],
    'CurrentAssetResidual': working_capital['CurrentAssetResidual'],
    'PPE_Net': ppe['PPE_Net'],
    'AP': working_capital['AP'],
    'AccruedExpenses': working_capital['AccruedExpenses'],
    'DeferredRevenue': working_capital['DeferredRevenue'],
    'OtherCurrentLiabilities': working_capital['OtherCurrentLiabilities'],
    'Debt': debt['Debt'],
    'TotalEquity': equity['TotalEquity'],
    'ReportedTotalAssets': spine['TotalAssets'],
    'ReportedTotalLiabilities': spine['TotalLiabilities'],
    'ReportedTotalEquity': spine['TotalEquity'],
})

#quantify the other
balance_sheet['OtherAssetResidual'] = (
    balance_sheet['ReportedTotalAssets']
    - balance_sheet['ReportedCash']
    - balance_sheet['ShortTermInvestments']
    - balance_sheet['AR']
    - balance_sheet['Inventory']
    - balance_sheet['OtherCurrentAssets']
    - balance_sheet['CurrentAssetResidual']
    - balance_sheet['PPE_Net']
)
#quantify the other again
balance_sheet['OtherLiabilityResidual'] = (
    balance_sheet['ReportedTotalLiabilities']
    - balance_sheet['AP']
    - balance_sheet['AccruedExpenses']
    - balance_sheet['DeferredRevenue']
    - balance_sheet['OtherCurrentLiabilities']
    - balance_sheet['Debt']
)
#quantify the TA excluding cash
balance_sheet['TotalAssets_exCash'] = (
    balance_sheet['ShortTermInvestments']
    + balance_sheet['AR']
    + balance_sheet['Inventory']
    + balance_sheet['OtherCurrentAssets']
    + balance_sheet['CurrentAssetResidual']
    + balance_sheet['PPE_Net']
    + balance_sheet['OtherAssetResidual']
)

#quantify TL
balance_sheet['TotalLiabilities'] = (
    balance_sheet['AP']
    + balance_sheet['AccruedExpenses']
    + balance_sheet['DeferredRevenue']
    + balance_sheet['OtherCurrentLiabilities']
    + balance_sheet['Debt']
    + balance_sheet['OtherLiabilityResidual']
)


balance_sheet['Cash_Reconciled'] = balance_sheet['TotalLiabilities'] + balance_sheet['TotalEquity'] - balance_sheet['TotalAssets_exCash']
balance_sheet['TotalAssets'] = balance_sheet['TotalAssets_exCash'] + balance_sheet['Cash_Reconciled']
balance_sheet['CashPlugGap'] = balance_sheet['Cash_Reconciled'] - balance_sheet['ReportedCash']
balance_sheet['AssetsGap'] = balance_sheet['TotalAssets'] - balance_sheet['ReportedTotalAssets']
balance_sheet['LiabilitiesGap'] = balance_sheet['TotalLiabilities'] - balance_sheet['ReportedTotalLiabilities']
balance_sheet['EquityGap'] = balance_sheet['TotalEquity'] - balance_sheet['ReportedTotalEquity']
balance_sheet['BalanceError'] = balance_sheet['TotalAssets'] - balance_sheet['TotalLiabilities'] - balance_sheet['TotalEquity']

#Balance check (A = L + E)
balance_check = pd.DataFrame({
    'Quarter': balance_sheet['Quarter'],
    'PeriodKey': balance_sheet['PeriodKey'],
    'Cash_Reconciled': balance_sheet['Cash_Reconciled'],
    'ReportedCash': balance_sheet['ReportedCash'],
    'CashPlugGap': balance_sheet['CashPlugGap'],
    'BalanceError': balance_sheet['BalanceError'],
    'AssetsGap': balance_sheet['AssetsGap'],
    'LiabilitiesGap': balance_sheet['LiabilitiesGap'],
    'EquityGap': balance_sheet['EquityGap'],
    'CCC': working_capital['CCC'],
    'WC_CashImpact_Gap': working_capital['WC_CashImpact_Gap'],
    'PPE_OtherAdjustments': ppe['OtherPPEAdjustments'],
    'CapitalActionsAndOCI': equity['CapitalActionsAndOCI'],
    'Debt_Change': debt['Debt_Change'],
})

#Run all the checks and display them to ensure correctness
balance_check['NegativeCashFlag'] = balance_sheet['Cash_Reconciled'] < 0.0
balance_check['NegativeARFlag'] = balance_sheet['AR'] < 0.0
balance_check['NegativeInventoryFlag'] = balance_sheet['Inventory'] < 0.0
balance_check['NegativeAPFlag'] = balance_sheet['AP'] < 0.0
balance_check['NegativePPEFlag'] = balance_sheet['PPE_Net'] < 0.0
balance_check['NegativeDebtFlag'] = balance_sheet['Debt'] < 0.0
balance_check['NegativeTotalEquityFlag'] = balance_sheet['TotalEquity'] < 0.0
balance_check['CCC_OutlierFlag'] = working_capital['CCC_OutlierFlag'].astype('bool')
balance_check['PPE_Adjustment_OutlierFlag'] = ppe['PPE_Adjustment_OutlierFlag'].astype('bool')
balance_check['Equity_Adjustment_OutlierFlag'] = equity['Equity_Adjustment_OutlierFlag'].astype('bool')
balance_check['DebtMovement_OutlierFlag'] = debt['DebtMovement_OutlierFlag'].astype('bool')
balance_check['CashPlugMismatchFlag'] = balance_check['CashPlugGap'].abs().gt(1e-2)
balance_check['BalanceErrorFlag'] = balance_check['BalanceError'].abs().gt(1e-2)
flag_cols = [col for col in balance_check.columns if col.endswith('Flag')]
balance_check['NegativeBalanceCount'] = balance_check[[col for col in flag_cols if col.startswith('Negative')]].sum(axis=1)
balance_check['OutlierCount'] = balance_check[[col for col in flag_cols if 'Outlier' in col]].sum(axis=1)
balance_check['FlagCount'] = balance_check[flag_cols].sum(axis=1)

working_capital_q = working_capital.set_index('Quarter')
ppe_q = ppe.set_index('Quarter')
equity_q = equity.set_index('Quarter')
debt_q = debt.set_index('Quarter')
balance_sheet_q = balance_sheet.set_index('Quarter')
balance_check_q = balance_check.set_index('Quarter')

#Outputs fro control pupose, using the function we defined in section 2 
output_files = {
    'working_capital_csv': write_schedule_csv(working_capital, 'chapter4_working_capital_schedule_usdm.csv'),
    'ppe_csv': write_schedule_csv(ppe, 'chapter4_ppe_schedule_usdm.csv'),
    'equity_csv': write_schedule_csv(equity, 'chapter4_equity_schedule_usdm.csv'),
    'debt_csv': write_schedule_csv(debt, 'chapter4_debt_schedule_usdm.csv'),
    'balance_sheet_csv': write_schedule_csv(balance_sheet, 'chapter4_balance_sheet_usdm.csv'),
    'balance_check_csv': write_schedule_csv(balance_check, 'chapter4_balance_check_report.csv'),
}


print('Schedules built and Chapter 4 outputs exported.')
print(f"Max abs balance error: {balance_check['BalanceError'].abs().max():,.6f}")
print(f"Max abs cash plug gap: {balance_check['CashPlugGap'].abs().max():,.6f}")

wc_driver_view = working_capital_q[
    [
        'Revenue', 'COGS', 'DSO', 'DIO', 'DPO', 'CCC', 'AR',
        'Inventory', 'AP', 'NWC', 'WC_CashImpact', 'ReportedCF_WorkingCapital',
        'WC_CashImpact_Gap'
    ]
]
display(wc_driver_view)

#median CCC to use in the executive summary

median_ccc = working_capital_q["CCC"].median()
display(median_ccc)

wc_support_view = working_capital_q[
    [
        'AR_Reported', 'AR_Error', 'Inventory_Reported', 'Inventory_Error',
        'AP_Reported', 'AP_Error', 'OtherCurrentAssets', 'CurrentAssetResidual',
        'AccruedExpenses', 'DeferredRevenue', 'OtherCurrentLiabilities',
        'CCC_OutlierFlag'
    ]
]
display(wc_support_view.head(12))

wc_summary = working_capital[['DSO', 'DIO', 'DPO', 'CCC', 'WC_CashImpact_Gap']].describe().T
print(f'CCC outlier quarters: {int(working_capital["CCC_OutlierFlag"].sum())}')
display(wc_summary)

ppe_view = ppe_q[
    [
        'PPE_Open', 'CapEx', 'Depreciation', 'OtherPPEAdjustments',
        'PPE_Net', 'PPE_Reported', 'PPE_Error', 'PPE_Adjustment_OutlierFlag'
    ]
]
display(ppe_view)

#For usage in the executive summary
largest_ppe_residual_period = ppe.loc[ppe['OtherPPEAdjustments'].abs().idxmax(), 'Quarter']
print(f'Largest PP&E residual quarter: {largest_ppe_residual_period}')
print(f"Largest abs PP&E residual: {ppe['OtherPPEAdjustments'].abs().max():,.2f} USDm")

equity_view = equity_q[
    [
        'PeriodKey', 'Equity_Open', 'NetIncome', 'Dividends',
        'StockBasedCompensation', 'RetainedEarningsLinkage',
        'CapitalActionsAndOCI', 'TotalEquity', 'TotalEquity_Reported',
        'TotalEquity_Error', 'Equity_Adjustment_OutlierFlag'
    ]
]
display(equity_view.head(12))

print(f'Equity residual outlier quarters: {int(equity["Equity_Adjustment_OutlierFlag"].sum())}')
print(f"Largest abs capital-actions / OCI residual: {equity['CapitalActionsAndOCI'].abs().max():,.2f} USDm")

debt_view = debt_q[
    [
        'Debt_Open', 'Borrowings', 'Repayments', 'Debt', 'Debt_Reported',
        'Debt_Error', 'Debt_Change', 'AverageDebt', 'InterestRate_Implied',
        'InterestExpense_Reported', 'InterestExpense_Modeled', 'DebtMovement_OutlierFlag'
    ]
]
display(debt_view)

print(f'Debt movement outlier quarters: {int(debt["DebtMovement_OutlierFlag"].sum())}')
print(f"Largest abs debt change: {debt['Debt_Change'].abs().max():,.2f} USDm")

balance_view = balance_sheet_q[
    [
        'ReportedCash', 'Cash_Reconciled', 'CashPlugGap', 'ShortTermInvestments',
        'AR', 'Inventory', 'OtherCurrentAssets', 'CurrentAssetResidual', 'OtherAssetResidual',
        'PPE_Net', 'TotalAssets', 'AP', 'AccruedExpenses', 'DeferredRevenue',
        'OtherCurrentLiabilities', 'Debt', 'OtherLiabilityResidual', 'TotalLiabilities',
        'TotalEquity', 'BalanceError'
    ]
]
display(balance_view.tail(8))
print(f"Max abs balance error: {balance_sheet['BalanceError'].abs().max():,.6f}")
print(f"Max abs cash plug gap: {balance_sheet['CashPlugGap'].abs().max():,.6f}")

display(balance_check_q)

flagged_periods = balance_check_q[balance_check_q['FlagCount'] > 0]
if flagged_periods.empty:
    print('No balance-check flags triggered.')
else:
    display(flagged_periods)

#asseting that max error can be 0.01
assert balance_check['BalanceError'].abs().max() < 1e-2
assert balance_check['CashPlugGap'].abs().max() < 1e-2
print('Verification passed: the reconstructed balance sheet balances and the reconciled cash line matches the reported cash balance.')