Unit 1 Mini-Project¶

# As mentioned, keeping the good practise of doing all the importations on top of the code
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
from pathlib import Path
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}")

# same we will always use to find root 
def find_project_root(start: Path | None = None) -> Path:
    start_path = (start or Path.cwd()).resolve()
    for candidate in [start_path, *start_path.parents]:
        if (candidate / "DATA" / "csv_export").exists() and (candidate / "Mini-projects").exists():
            return candidate
    raise FileNotFoundError("Could not find the project root containing DATA/csv_export and Mini-projects.")

# Function for reading the csv
def load_source_csv(csv_dir: Path, filename: str) -> pd.DataFrame:
    path = csv_dir / filename
    header = pd.read_csv(path, nrows=0)
    parse_dates = [c for c in ["date", "acceptedDate", "filingDate"] if c in header.columns]
    return pd.read_csv(path, parse_dates=parse_dates)

# use the root function + actual data directions
PROJECT_ROOT = find_project_root()
CSV_DIR = PROJECT_ROOT / "DATA" / "csv_export"
OUT_DIR = PROJECT_ROOT / "Mini-projects" / "outputs"
QUARTERLY_EXPORT_DIR = CSV_DIR

# Loading csv
is_raw = load_source_csv(CSV_DIR, "RL_income_statement.csv")
bs_raw = load_source_csv(CSV_DIR, "RL_balance_sheet.csv")
cf_raw = load_source_csv(CSV_DIR, "RL_cash_flow.csv")
ratios_raw = load_source_csv(CSV_DIR, "RL_ratios.csv")

# Printing the shape information, since we have 40 quarters of data should always be (40,n)
print("Project root:", PROJECT_ROOT)
print("Loaded tables:")
print("  Income statement:", is_raw.shape)
print("  Balance sheet   :", bs_raw.shape)
print("  Cash flow       :", cf_raw.shape)
print("  Ratios          :", ratios_raw.shape)

# Using what we just created and bringing in the raw data
tidy, audit = build_tidy_pack(is_raw, bs_raw, cf_raw, ratios_raw)

print("Fiscal quarters in tidy dataset:", tidy.index[0], "to", tidy.index[-1])
print("Rows x columns:", tidy.shape)
display(tidy.head(8))
display(audit.head(8))

# Wasn't required but for control purpose we are going to save a tidy CSV that can be loaded to other projects data that is already ready to go.
out_dir = OUT_DIR
out_dir.mkdir(parents=True, exist_ok=True)
tidy.to_csv(out_dir / "unit1_tidy_historical_pack_usdm.csv")
audit.to_csv(out_dir / "unit1_tidy_historical_audit.csv")

#Defining the 2 main functions that we are going to use for conduct this analysis

def common_size(df: pd.DataFrame, base_col: str, cols: list[str]) -> pd.DataFrame:
    base = pd.to_numeric(df[base_col], errors="coerce").replace(0, np.nan)
    out = pd.DataFrame(index=df.index)
    for c in cols:
        if c in df.columns:
            out[c] = pd.to_numeric(df[c], errors="coerce") / base
    return out * 100.0


def add_trend_metrics(df: pd.DataFrame, cols: list[str]) -> pd.DataFrame:
    out = df.copy()
    for c in cols:
        series = pd.to_numeric(out[c], errors="coerce")
        out[c] = series
        out[f"{c}_QoQ_pct"] = series.pct_change() * 100
        out[f"{c}_YoY_pct"] = series.pct_change(4) * 100
        out[f"{c}_TTM"] = series.rolling(4, min_periods=4).sum()
    return out

# Settings columns with information that we are going to analyse
is_cs_cols = ["Revenue", "COGS", "GrossProfit", "OperatingExpenses", "OperatingIncome", "NetIncome"]
bs_cs_cols = ["Cash", "AccountsReceivable", "Inventory", "PPENet", "TotalDebt", "TotalLiabilities", "EquityForBalance", "TotalAssets"]

common_is = common_size(tidy, "Revenue", [c for c in is_cs_cols if c in tidy.columns])
common_bs = common_size(tidy, "TotalAssets", [c for c in bs_cs_cols if c in tidy.columns])

# Creating columns for the trending analyse
trend_cols = [c for c in ["Revenue", "GrossProfit", "OperatingIncome", "NetIncome", "OperatingCashFlow", "FreeCashFlow", "CapEx"] if c in tidy.columns]
trend = add_trend_metrics(tidy[trend_cols], trend_cols)

display(common_is.round(2))
display(common_bs.round(2))
display(trend.round(2))

fig, axes = plt.subplots(1, 2, figsize=(16, 5))

marg = pd.DataFrame(index=tidy.index)
if {"GrossProfit", "Revenue"}.issubset(tidy.columns):
    marg["GrossMargin"] = tidy["GrossProfit"] / tidy["Revenue"].replace(0, np.nan) * 100
if {"OperatingIncome", "Revenue"}.issubset(tidy.columns):
    marg["OperatingMargin"] = tidy["OperatingIncome"] / tidy["Revenue"].replace(0, np.nan) * 100
if {"NetIncome", "Revenue"}.issubset(tidy.columns):
    marg["NetMargin"] = tidy["NetIncome"] / tidy["Revenue"].replace(0, np.nan) * 100

for col in marg.columns:
    axes[0].plot(marg.index, marg[col], marker="o", linewidth=1.8, label=col)
axes[0].set_title("Quarterly Common-Size Margins (%)")
axes[0].set_xlabel("Fiscal Quarter")
axes[0].set_ylabel("Percent")
axes[0].tick_params(axis="x", rotation=45)
axes[0].legend()

#Formating for getting plot ready
plot_cols = [c for c in ["Revenue", "NetIncome"] if c in tidy.columns]
for col in plot_cols:
    axes[1].plot(tidy.index, tidy[col], marker="o", linewidth=1.8, label=col)
axes[1].set_title("Quarterly Level Trends (USD m)")
axes[1].set_xlabel("Fiscal Quarter")
axes[1].set_ylabel("USD m")
axes[1].tick_params(axis="x", rotation=45)
axes[1].legend()

#plotting
plt.tight_layout()
plt.show()

def _avg(s: pd.Series) -> pd.Series:
    return (s + s.shift(1)) / 2


# defining the main fuction to create the ratio matrix and making sure we are covering all families and have no blank datas
def build_ratio_matrix(tidy_df: pd.DataFrame) -> pd.DataFrame:
    d = tidy_df.copy()
    r = pd.DataFrame(index=d.index)

    if {"TotalCurrentAssets", "TotalCurrentLiabilities"}.issubset(d.columns):
        r["Liquidity_CurrentRatio"] = d["TotalCurrentAssets"] / d["TotalCurrentLiabilities"].replace(0, np.nan)
    if {"Cash", "AccountsReceivable", "TotalCurrentLiabilities"}.issubset(d.columns):
        r["Liquidity_QuickRatio"] = (d["Cash"] + d["AccountsReceivable"]) / d["TotalCurrentLiabilities"].replace(0, np.nan)
    if {"Cash", "TotalCurrentLiabilities"}.issubset(d.columns):
        r["Liquidity_CashRatio"] = d["Cash"] / d["TotalCurrentLiabilities"].replace(0, np.nan)

    if {"GrossProfit", "Revenue"}.issubset(d.columns):
        r["Profitability_GrossMargin"] = d["GrossProfit"] / d["Revenue"].replace(0, np.nan)
    if {"EBIT", "Revenue"}.issubset(d.columns):
        r["Profitability_EBITMargin"] = d["EBIT"] / d["Revenue"].replace(0, np.nan)
    if {"NetIncome", "Revenue"}.issubset(d.columns):
        r["Profitability_NetMargin"] = d["NetIncome"] / d["Revenue"].replace(0, np.nan)
    if {"NetIncome_TTM", "AvgAssets"}.issubset(d.columns):
        r["Profitability_ROA"] = d["NetIncome_TTM"] / d["AvgAssets"].replace(0, np.nan)
    if {"NetIncome_TTM", "AvgEquity"}.issubset(d.columns):
        r["Profitability_ROE"] = d["NetIncome_TTM"] / d["AvgEquity"].replace(0, np.nan)

    if {"AvgAR", "Revenue_TTM"}.issubset(d.columns):
        r["Efficiency_ARDays"] = d["AvgAR"] / (d["Revenue_TTM"] / 365).replace(0, np.nan)
    if {"AvgInventory", "COGS_TTM"}.issubset(d.columns):
        r["Efficiency_InventoryDays"] = d["AvgInventory"] / (d["COGS_TTM"] / 365).replace(0, np.nan)
    if {"AvgAP", "COGS_TTM"}.issubset(d.columns):
        r["Efficiency_APDays"] = d["AvgAP"] / (d["COGS_TTM"] / 365).replace(0, np.nan)
    if {"Efficiency_ARDays", "Efficiency_InventoryDays", "Efficiency_APDays"}.issubset(r.columns):
        r["Efficiency_CCC"] = r["Efficiency_ARDays"] + r["Efficiency_InventoryDays"] - r["Efficiency_APDays"]
    if {"Revenue_TTM", "AvgAssets"}.issubset(d.columns):
        r["Efficiency_AssetTurnover"] = d["Revenue_TTM"] / d["AvgAssets"].replace(0, np.nan)

    if {"TotalDebt", "EBITDA_TTM"}.issubset(d.columns):
        r["Leverage_DebtToEBITDA"] = d["TotalDebt"] / d["EBITDA_TTM"].replace(0, np.nan)
    if {"TotalDebt", "EquityForBalance"}.issubset(d.columns):
        r["Leverage_DebtToEquity"] = d["TotalDebt"] / d["EquityForBalance"].replace(0, np.nan)
    if {"TotalDebt", "TotalAssets"}.issubset(d.columns):
        r["Leverage_DebtToAssets"] = d["TotalDebt"] / d["TotalAssets"].replace(0, np.nan)
    if {"NetDebt", "EBITDA_TTM"}.issubset(d.columns):
        r["Leverage_NetDebtToEBITDA"] = d["NetDebt"] / d["EBITDA_TTM"].replace(0, np.nan)
    if {"EBIT_TTM", "InterestExpense_TTM"}.issubset(d.columns):
        r["Leverage_InterestCoverage"] = d["EBIT_TTM"] / d["InterestExpense_TTM"].replace(0, np.nan)

    for col in ["PE", "PS", "PB", "EV_EBITDA", "DebtToMarketCap"]:
        if col in d.columns:
            r[f"Valuation_{col}"] = d[col]

    return r.replace([np.inf, -np.inf], np.nan)


# Using the function in the tidy dataframe
ratio_matrix = build_ratio_matrix(tidy)
display(ratio_matrix.round(3))

# Defining the rations that lower is better.
heat = ratio_matrix.copy()
lower_is_better = [
    "Efficiency_ARDays", "Efficiency_InventoryDays", "Efficiency_APDays", "Efficiency_CCC",
    "Leverage_DebtToEBITDA", "Leverage_DebtToEquity", "Leverage_DebtToAssets", "Leverage_NetDebtToEBITDA",
    "Valuation_PE", "Valuation_PS", "Valuation_PB", "Valuation_EV_EBITDA", "Valuation_DebtToMarketCap",
]
for col in lower_is_better:
    if col in heat.columns:
        heat[col] = -heat[col]

z = (heat - heat.mean()) / heat.std(ddof=0)
z = z.replace([np.inf, -np.inf], np.nan)
z_plot = z.dropna(axis=1, how="all").T

# Plotting
fig, ax = plt.subplots(figsize=(16, 7))
im = ax.imshow(z_plot.values, aspect="auto")
ax.set_yticks(range(len(z_plot.index)))
ax.set_yticklabels(z_plot.index)
ax.set_xticks(range(len(z_plot.columns)))
ax.set_xticklabels(z_plot.columns, rotation=45, ha="right")
ax.set_title("Quarterly Ratio Dashboard Heat Map (within-ratio z-scores)")
ax.set_xlabel("Fiscal Quarter")
ax.set_ylabel("Ratio")
plt.colorbar(im, ax=ax, fraction=0.046, pad=0.04)
plt.tight_layout()
plt.show()

# Defining the buffers
def build_activation_red_flags(
    ratios: pd.DataFrame,
    rolling_window: int | None = None,
    min_periods: int = 3,
    liquidity_buffer: float = 0.1,
    leverage_buffer: float = 0.15,
    coverage_buffer: float = 0.15,
    operating_buffer: float = 0.1,
    ccc_jump_threshold: float = 20.0,
    wc_buffer: float = 0.2,
) -> pd.DataFrame:
    
    r = ratios.copy()
    flags = pd.DataFrame(index=r.index)

    if rolling_window is None:
        rolling_window = 12 if len(r) >= 12 else max(3, len(r) // 2)
    mp = min(min_periods, rolling_window)

    def _get_col(candidates: list[str]) -> pd.Series | None:
        for c in candidates:
            if c in r.columns:
                return pd.to_numeric(r[c], errors="coerce")
        return None

    def _rm(s: pd.Series) -> pd.Series:
        return s.rolling(rolling_window, min_periods=mp).mean()

    def _or_reduce(conds: list[pd.Series]) -> pd.Series:
        if not conds:
            return pd.Series(False, index=r.index)
        arr = [c.fillna(False).astype(bool) for c in conds]
        return pd.Series(np.logical_or.reduce(arr), index=r.index)

    liquidity_conditions = []

    quick = _get_col(["QuickRatio", "Liquidity_QuickRatio"])
    if quick is not None:
        quick_rm = _rm(quick)
        liquidity_conditions.append(quick < quick_rm * (1 - liquidity_buffer))

    current = _get_col(["CurrentRatio", "Liquidity_CurrentRatio"])
    if current is not None:
        current_rm = _rm(current)
        liquidity_conditions.append(current < current_rm * (1 - liquidity_buffer))

    cash = _get_col(["CashRatio", "Liquidity_CashRatio"])
    if cash is not None:
        cash_rm = _rm(cash)
        liquidity_conditions.append(cash < cash_rm * (1 - liquidity_buffer))

    flags["RF1_LiquidityStress"] = _or_reduce(liquidity_conditions)

    # RF2: Leverage Risk
    leverage_conditions = []

    debt_ebitda = _get_col(["DebtEBITDA", "Leverage_DebtToEBITDA"])
    if debt_ebitda is not None:
        debt_ebitda_rm = _rm(debt_ebitda)
        leverage_conditions.append(debt_ebitda > debt_ebitda_rm * (1 + leverage_buffer))

    net_debt_ebitda = _get_col(["NetDebtEBITDA", "Leverage_NetDebtToEBITDA"])
    if net_debt_ebitda is not None:
        nde_rm = _rm(net_debt_ebitda)
        leverage_conditions.append(net_debt_ebitda > nde_rm * (1 + leverage_buffer))

    debt_to_equity = _get_col(["DebtToEquity", "Leverage_DebtToEquity"])
    if debt_to_equity is not None:
        dte_rm = _rm(debt_to_equity)
        leverage_conditions.append(debt_to_equity > dte_rm * (1 + leverage_buffer))

    debt_to_assets = _get_col(["DebtToAssets", "Leverage_DebtToAssets"])
    if debt_to_assets is not None:
        dta_rm = _rm(debt_to_assets)
        leverage_conditions.append(debt_to_assets > dta_rm * (1 + leverage_buffer))

    int_cov = _get_col(["IntCoverage", "Leverage_InterestCoverage"])
    if int_cov is not None:
        int_cov_rm = _rm(int_cov)
        leverage_conditions.append(int_cov < int_cov_rm * (1 - coverage_buffer))

    flags["RF2_LeverageRisk"] = _or_reduce(leverage_conditions)

    # RF3: Operating Deterioration
    op_conditions = []
    for candidates in [["GrossMargin", "Profitability_GrossMargin"],
                       ["EBITMargin", "Profitability_EBITMargin"],
                       ["ROA", "Profitability_ROA"]]:
        s = _get_col(candidates)
        if s is None:
            continue
        s_rm = _rm(s)
        op_conditions.append((s < s_rm * (1 - operating_buffer)) & (s.diff() < 0))

    flags["RF3_OperatingDeterioration"] = _or_reduce(op_conditions)

    # RF4: Working Capital Shock
    wc_conditions = []

    ccc = _get_col(["CCC", "Efficiency_CCC"])
    if ccc is not None:
        ccc_rm = _rm(ccc)
        wc_conditions.append((ccc.diff() > ccc_jump_threshold) | (ccc > ccc_rm * (1 + wc_buffer)))

    inv_days = _get_col(["InventoryDays", "Efficiency_InventoryDays"])
    if inv_days is not None:
        inv_rm = _rm(inv_days)
        wc_conditions.append((inv_days > inv_rm * (1 + wc_buffer)) & (inv_days.diff() > 0))

    ar_days = _get_col(["ARDays", "Efficiency_ARDays"])
    if ar_days is not None:
        ar_rm = _rm(ar_days)
        wc_conditions.append((ar_days > ar_rm * (1 + wc_buffer)) & (ar_days.diff() > 0))

    flags["RF4_WorkingCapitalShock"] = _or_reduce(wc_conditions)

    return flags


flags = build_activation_red_flags(ratio_matrix)
# Making the Redflag text and identification
driver_text = {
    "RF1_LiquidityStress": "Liquidity ratios are below buffered rolling norms, indicating tighter near-term funding flexibility.",
    "RF2_LeverageRisk": "Leverage metrics are above buffered rolling norms and/or coverage is below buffered norms.",
    "RF3_OperatingDeterioration": "Core operating metrics are below buffered rolling trend and still moving down.",
    "RF4_WorkingCapitalShock": "CCC and/or working-capital days are rising above buffered rolling baseline (cash conversion drag).",
}

# Activating the flags
activated_flags = (
    flags[flags.any(axis=1)]
    .stack()
    .reset_index()
    .rename(columns={"level_0": "Year", "level_1": "Rule", 0: "Triggered"})
)
activated_flags = activated_flags[activated_flags["Triggered"]].drop(columns="Triggered")
if len(activated_flags) > 0:
    activated_flags["Driver"] = activated_flags["Rule"].map(driver_text)

# printing the outputs
print("Flag matrix (rolling-mean + buffer rules):")
display(flags)

print()
print("Activated flags with drivers:")
if len(activated_flags) == 0:
    print("No flags triggered under current thresholds.")
else:
    display(activated_flags)
    
# Created a flag count to help us in the executive summary
flag_counts = flags.sum(axis=1).rename("RedFlagCount")
display(flag_counts.to_frame())
x = flag_counts.sum()
print(f"Total Red flags activated: {x}")