Практическая задача

Ты — senior Python data analyst, finance analyst и AI automation engineer.

Напиши полностью готовый код для Google Colab для анализа и прогнозирования кассовых разрывов компании.

ЗАДАЧА
Необходимо считать данные по движению денежных средств из Google Sheets или Excel, провести анализ cash flow, выявить периоды кассового разрыва, построить прогноз и сформировать графический аналитический отчет.

ИСТОЧНИК ДАННЫХ
Google Sheets / Excel

ОСНОВНЫЕ ПОЛЯ

• date

• opening_balance_kzt

• inflow_kzt

• outflow_kzt

• closing_balance_kzt

ТРЕБУЕТСЯ

1. считать данные

2. очистить данные

3. рассчитать net cash flow

4. выявить отрицательные остатки

5. определить даты кассового разрыва

6. построить прогноз на 30, 60 и 90 дней

7. построить графики движения денежных средств

8. сформировать аналитическую таблицу

9. выгрузить отчет в Excel

10. подготовить краткие выводы и рекомендации

ВИЗУАЛИЗАЦИИ

• график остатка денежных средств

• график поступлений и выплат

• выделение gap periods

• прогноз с доверительным интервалом

• dashboard KPI

ФОРМАТ ВЫВОДА
Сначала кратко опиши логику решения, затем выдай полный рабочий код для Google Colab.

# =========================================================

# EXECUTIVE CASH FLOW ANALYSIS PROGRAM

# Google Colab

# Файл: cashflow_analysis_dataset.xlsx

# Лист: первый лист файла

# =========================================================

# =========================

# 1. Установка библиотек

# =========================

!pip -q install openpyxl prophet plotly

# =========================

# 2. Импорт библиотек

# =========================

import pandas as pd

import numpy as np

import matplotlib.pyplot as plt

import plotly.graph_objects as go

import plotly.express as px

from prophet import Prophet

from google.colab import files

# =========================

# 3. Загрузка файла

# =========================

uploaded = files.upload()

FILE_NAME = "cashflow_analysis_dataset.xlsx"

xls = pd.ExcelFile(FILE_NAME)

print("Доступные листы в файле:")

print(xls.sheet_names)

df = pd.read_excel(FILE_NAME, sheet_name=xls.sheet_names[0])

print("Данные успешно загружены")

display(df.head())

# =========================

# 4. Приведение русских названий столбцов к рабочему формату

# =========================

df = df.rename(columns={

   "Дата": "date",

   "Входящий остаток, тг": "opening_balance_kzt",

   "Поступления от продаж, тг": "inflow_sales_kzt",

   "Погашение дебиторской задолженности, тг": "inflow_collections_kzt",

   "Прочие поступления, тг": "inflow_other_kzt",

   "Общие поступления, тг": "inflow_kzt",

   "Оплата поставщикам, тг": "outflow_suppliers_kzt",

   "Фонд оплаты труда, тг": "outflow_payroll_kzt",

   "Налоги и обязательные платежи, тг": "outflow_tax_kzt",

   "Арендные платежи, тг": "outflow_rent_kzt",

   "Погашение кредита, тг": "outflow_loan_kzt",

   "Прочие выплаты, тг": "outflow_other_kzt",

   "Общие выплаты, тг": "outflow_kzt",

   "Исходящий остаток, тг": "closing_balance_kzt"

})

print("Названия столбцов приведены к формату программы")

display(df.head())

# =========================

# 5. Подготовка данных

# =========================

df["date"] = pd.to_datetime(df["date"], errors="coerce")

df = df.sort_values("date").reset_index(drop=True)

numeric_cols = [c for c in df.columns if c != "date"]

for col in numeric_cols:

   df[col] = pd.to_numeric(df[col], errors="coerce")

df = df.dropna(subset=["date"])

# Основные расчеты

df["net_cashflow_kzt"] = df["inflow_kzt"] - df["outflow_kzt"]

df["cash_gap_flag"] = np.where(df["closing_balance_kzt"] < 0, 1, 0)

df["gap_amount_kzt"] = np.where(df["closing_balance_kzt"] < 0, -df["closing_balance_kzt"], 0)

# Скользящие показатели

df["rolling_inflow_7d"] = df["inflow_kzt"].rolling(7, min_periods=1).mean()

df["rolling_outflow_7d"] = df["outflow_kzt"].rolling(7, min_periods=1).mean()

df["days_of_cash"] = np.where(

   df["rolling_outflow_7d"] > 0,

   df["closing_balance_kzt"] / df["rolling_outflow_7d"],

   np.nan

)

df["days_of_cash"] = df["days_of_cash"].replace([np.inf, -np.inf], np.nan).fillna(0)

df["risk_flag"] = np.where(

   (df["closing_balance_kzt"] < 0) | (df["days_of_cash"] < 3),

   1,

   0

)

df["month_name"] = df["date"].dt.strftime("%Y-%m")

print("Подготовка данных завершена")

display(df.head())

# =========================

# 6. KPI summary

# =========================

total_inflow = df["inflow_kzt"].sum()

total_outflow = df["outflow_kzt"].sum()

average_balance = df["closing_balance_kzt"].mean()

minimum_balance = df["closing_balance_kzt"].min()

maximum_gap = df["gap_amount_kzt"].max()

gap_days = int(df["cash_gap_flag"].sum())

avg_days_of_cash = df["days_of_cash"].mean()

risk_share = df["risk_flag"].mean() * 100

# =========================

# 7. KPI Dashboard

# =========================

fig_kpi = go.Figure()

kpi_numeric_values = [

   total_inflow,

   total_outflow,

   average_balance,

   minimum_balance,

   gap_days,

   maximum_gap,

   avg_days_of_cash,

   risk_share

]

kpi_titles = [

   "Общие поступления",

   "Общие выплаты",

   "Средний остаток",

   "Минимальный остаток",

   "Дни разрыва",

   "Максимальный разрыв",

   "Средний запас ликвидности",

   "Доля риск-дней"

]

kpi_suffix = [

   " тг",

   " тг",

   " тг",

   " тг",

   "",

   " тг",

   "",

   "%"

]

for i, (title, val, suffix) in enumerate(zip(kpi_titles, kpi_numeric_values, kpi_suffix)):

   fig_kpi.add_trace(go.Indicator(

       mode="number",

       value=val,

       number={"suffix": suffix},

       title={"text": f"<b>{title}</b>"},

       domain={"row": i // 4, "column": i % 4}

   ))

fig_kpi.update_layout(

   grid={"rows": 2, "columns": 4, "pattern": "independent"},

   title="KPI Dashboard — Анализ cash flow",

   height=500

)

fig_kpi.show()

# =========================

# 8. Основные графики

# =========================

# 8.1 Динамика остатка денежных средств

plt.figure(figsize=(14, 6))

plt.plot(df["date"], df["closing_balance_kzt"], linewidth=2, label="Исходящий остаток")

plt.axhline(0, linestyle="--", color="red", label="Нулевая линия")

plt.title("Динамика остатка денежных средств")

plt.xlabel("Дата")

plt.ylabel("тенге")

plt.legend()

plt.grid(True)

plt.tight_layout()

plt.show()

# 8.2 Поступления и выплаты

plt.figure(figsize=(14, 6))

plt.plot(df["date"], df["inflow_kzt"], linewidth=2, label="Поступления")

plt.plot(df["date"], df["outflow_kzt"], linewidth=2, label="Выплаты")

plt.title("Поступления и выплаты")

plt.xlabel("Дата")

plt.ylabel("тенге")

plt.legend()

plt.grid(True)

plt.tight_layout()

plt.show()

# 8.3 Периоды кассовых разрывов

plt.figure(figsize=(14, 6))

gap_df = df[df["cash_gap_flag"] == 1]

plt.plot(df["date"], df["closing_balance_kzt"], linewidth=2, label="Исходящий остаток")

plt.scatter(gap_df["date"], gap_df["closing_balance_kzt"], s=90, label="Кассовый разрыв")

plt.axhline(0, linestyle="--", color="red")

plt.title("Периоды кассовых разрывов")

plt.xlabel("Дата")

plt.ylabel("тенге")

plt.legend()

plt.grid(True)

plt.tight_layout()

plt.show()

# =========================

# 9. Структура поступлений и выплат

# =========================

inflow_articles = [c for c in ["inflow_sales_kzt", "inflow_collections_kzt", "inflow_other_kzt"] if c in df.columns]

outflow_articles = [c for c in [

   "outflow_suppliers_kzt",

   "outflow_payroll_kzt",

   "outflow_tax_kzt",

   "outflow_rent_kzt",

   "outflow_loan_kzt",

   "outflow_other_kzt"

] if c in df.columns]

if inflow_articles:

   inflow_structure = pd.DataFrame({

       "Статья": inflow_articles,

       "Сумма, тг": [df[c].sum() for c in inflow_articles]

   })

   inflow_structure["Статья"] = inflow_structure["Статья"].replace({

       "inflow_sales_kzt": "Поступления от продаж",

       "inflow_collections_kzt": "Погашение дебиторской задолженности",

       "inflow_other_kzt": "Прочие поступления"

   })

   fig_inflow = px.pie(

       inflow_structure,

       values="Сумма, тг",

       names="Статья",

       title="Структура поступлений"

   )

   fig_inflow.show()

if outflow_articles:

   outflow_structure = pd.DataFrame({

       "Статья": outflow_articles,

       "Сумма, тг": [df[c].sum() for c in outflow_articles]

   })

   outflow_structure["Статья"] = outflow_structure["Статья"].replace({

       "outflow_suppliers_kzt": "Оплата поставщикам",

       "outflow_payroll_kzt": "Фонд оплаты труда",

       "outflow_tax_kzt": "Налоги и обязательные платежи",

       "outflow_rent_kzt": "Арендные платежи",

       "outflow_loan_kzt": "Погашение кредита",

       "outflow_other_kzt": "Прочие выплаты"

   })

   fig_outflow = px.pie(

       outflow_structure,

       values="Сумма, тг",

       names="Статья",

       title="Структура выплат"

   )

   fig_outflow.show()

# =========================

# 10. Ежемесячная сводка

# =========================

monthly_summary = (

   df.groupby("month_name", as_index=False)

   .agg(

       total_inflow_kzt=("inflow_kzt", "sum"),

       total_outflow_kzt=("outflow_kzt", "sum"),

       net_cashflow_kzt=("net_cashflow_kzt", "sum"),

       closing_balance_kzt=("closing_balance_kzt", "last"),

       gap_days=("cash_gap_flag", "sum"),

       max_gap_kzt=("gap_amount_kzt", "max")

   )

)

monthly_summary = monthly_summary.rename(columns={

   "month_name": "Период",

   "total_inflow_kzt": "Общие поступления, тг",

   "total_outflow_kzt": "Общие выплаты, тг",

   "net_cashflow_kzt": "Чистый денежный поток, тг",

   "closing_balance_kzt": "Исходящий остаток, тг",

   "gap_days": "Дни разрыва",

   "max_gap_kzt": "Максимальный разрыв, тг"

})

print("Ежемесячная сводка")

display(monthly_summary)

# =========================

# 11. Топ-10 дней с наибольшим разрывом

# =========================

top_gap_days = (

   df[df["gap_amount_kzt"] > 0]

   .sort_values("gap_amount_kzt", ascending=False)

   .head(10)

   .copy()

)

top_gap_days_display = top_gap_days[[

   "date",

   "closing_balance_kzt",

   "gap_amount_kzt",

   "inflow_kzt",

   "outflow_kzt"

]].rename(columns={

   "date": "Дата",

   "closing_balance_kzt": "Исходящий остаток, тг",

   "gap_amount_kzt": "Сумма разрыва, тг",

   "inflow_kzt": "Поступления, тг",

   "outflow_kzt": "Выплаты, тг"

})

print("Топ-10 дней с наибольшим кассовым разрывом")

display(top_gap_days_display)

# =========================

# 12. Прогноз на 30 дней

# =========================

forecast_input = df[["date", "closing_balance_kzt"]].rename(

   columns={"date": "ds", "closing_balance_kzt": "y"}

)

model = Prophet(interval_width=0.95)

model.fit(forecast_input)

future = model.make_future_dataframe(periods=30)

forecast = model.predict(future)

fig_forecast = model.plot(forecast)

plt.title("Прогноз остатка денежных средств на 30 дней")

plt.tight_layout()

plt.show()

# =========================

# 13. Executive summary

# =========================

future_only = forecast[forecast["ds"] > df["date"].max()].copy()

future_gap = future_only[future_only["yhat"] < 0]

next_gap_date = future_gap["ds"].min() if not future_gap.empty else None

recommended_reserve = df["rolling_outflow_7d"].mean() * 7

summary_text = f"""

EXECUTIVE SUMMARY

=================

Общие поступления: {total_inflow:,.0f} тг

Общие выплаты: {total_outflow:,.0f} тг

Средний остаток: {average_balance:,.0f} тг

Минимальный остаток: {minimum_balance:,.0f} тг

Количество дней с кассовым разрывом: {gap_days}

Максимальный кассовый разрыв: {maximum_gap:,.0f} тг

Среднее значение days of cash: {avg_days_of_cash:.2f}

Доля риск-дней: {risk_share:.1f}%

Следующий прогнозируемый кассовый разрыв:

{str(next_gap_date.date()) if next_gap_date is not None else "В горизонте 30 дней не прогнозируется"}

Рекомендуемый минимальный резерв ликвидности:

{recommended_reserve:,.0f} тг

Рекомендации:

1. Усилить контроль сроков поступления денежных средств.

2. Снизить концентрацию крупных выплат в одном периоде.

3. Пересмотреть график ФОТ, налогов и платежей поставщикам.

4. Использовать ежедневный dashboard для мониторинга ликвидности.

5. Применять сценарное планирование cash flow.

"""

print(summary_text)

# =========================

# 14. Выгрузка в Excel

# =========================

output_file = "cashflow_analysis_report.xlsx"

summary_df = pd.DataFrame({"Executive Summary": summary_text.split("\n")})

with pd.ExcelWriter(output_file, engine="openpyxl") as writer:

   df.to_excel(writer, sheet_name="analysis", index=False)

   monthly_summary.to_excel(writer, sheet_name="monthly_summary", index=False)

   top_gap_days_display.to_excel(writer, sheet_name="top_gap_days", index=False)

   forecast.to_excel(writer, sheet_name="forecast_30d", index=False)

   summary_df.to_excel(writer, sheet_name="executive_summary", index=False)

files.download(output_file)

print("Отчет успешно выгружен")

fig_kpi = go.Figure()

kpi_numeric_values = [

    total_inflow,

    total_outflow,

    average_balance,

    minimum_balance,

    gap_days,

    maximum_gap,

    avg_days_of_cash,

    risk_share

]

kpi_titles = [

    "Общие поступления",

    "Общие выплаты",

    "Средний остаток",

    "Мин. остаток",

    "Дни разрыва",

    "Макс. разрыв",

    "Запас ликвидности",

    "Риск-дни %"

]

kpi_suffix = [

    " тг",

    " тг",

    " тг",

    " тг",

    "",

    " тг",

    "",

    "%"

]

for i, (title, val, suffix) in enumerate(zip(kpi_titles, kpi_numeric_values, kpi_suffix)):

    fig_kpi.add_trace(go.Indicator(

        mode="number",

        value=val,

        number={

            "suffix": suffix,

            "font": {"size": 28}

        },

        title={

            "text": f"<b>{title}</b>",

            "font": {"size": 16}

        },

        domain={"row": i // 4, "column": i % 4}

    ))

fig_kpi.update_layout(

    grid={"rows": 2, "columns": 4, "pattern": "independent"},

    title={

        "text": "KPI Dashboard — Анализ cash flow",

        "font": {"size": 22}

    },

    height=420

)

fig_kpi.show()