tags, strip it and return chart_data."""
raw = text.strip()
if not raw:
return None, text
# Markdown fenced block
fence = re.search(r"```(?:json)?\s*(\{[\s\S]*?\})\s*```", raw, re.DOTALL)
if fence:
try:
d = json.loads(fence.group(1))
if _looks_like_forecast_chart_json(d):
clean = raw.replace(fence.group(0), "").strip()
return d, re.sub(r"\n{3,}", "\n\n", clean)
except Exception:
pass
# Whole message is just JSON
if raw.startswith("{") and raw.endswith("}"):
try:
d = json.loads(raw)
if _looks_like_forecast_chart_json(d):
return d, ""
except Exception:
pass
# Scan for embedded chart object (common when prose + JSON)
search_from = 0
while True:
blob, end = _extract_first_brace_json_object(raw, search_from)
if not blob:
break
try:
d = json.loads(blob)
if _looks_like_forecast_chart_json(d):
pos = raw.find(blob)
clean = (raw[:pos] + raw[pos + len(blob) :]).strip()
return d, re.sub(r"\n{3,}", "\n\n", clean)
except Exception:
pass
search_from = raw.find("{", search_from + 1)
if search_from < 0:
break
return None, text
def parse_and_render_chart(response_text: str) -> tuple[str, dict | None]:
"""Extract a โฆ block from an AI response.
Returns (clean_text, chart_data_dict). chart_data_dict is None when no
valid block is found.
"""
chart_pattern = r"(.*?)"
match = re.search(chart_pattern, response_text, re.DOTALL | re.IGNORECASE)
if match:
clean_text = re.sub(chart_pattern, "", response_text, flags=re.DOTALL | re.IGNORECASE).strip()
try:
chart_data = json.loads(match.group(1).strip())
return clean_text, chart_data
except Exception:
return clean_text, None
bare, stripped = _try_extract_bare_chart_json(response_text)
return stripped, bare
def _format_kwacha(v: float) -> str:
return f"K{v:,.0f}"
def _format_pct(v: float) -> str:
return f"{v:+.1f}%" if v else "0.0%"
def _caption_for_forecast_chart(chart_data: dict, scenario_label: str | None = None) -> str:
"""Generate a multi-sentence breakdown from a chart JSON.
Used both when the model returns no prose and when its prose is too thin
(so the user always gets a numbered breakdown alongside the chart).
"""
try:
title = (chart_data.get("title") or "").strip()
data = chart_data.get("data") or {}
labels = data.get("labels") or []
datasets = data.get("datasets") or []
if not labels or not datasets:
return ""
first, last = labels[0], labels[-1]
scen_suffix = f" under the **{scenario_label}** scenario" if scenario_label else ""
def _delta_pct(v0: float, vN: float) -> str:
if v0 == 0:
return ""
pct = (vN - v0) / abs(v0) * 100
return f" ({pct:+.1f}%)"
# Single-series shortcuts โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ
if len(datasets) == 1:
d = datasets[0]
name = (d.get("name") or "").strip() or title or "Series"
vals = d.get("values") or []
if not vals:
return ""
is_pct = "%" in name or "margin" in name.lower()
v0, vN = vals[0], vals[-1]
if is_pct:
trend = "improves" if vN > v0 else ("declines" if vN < v0 else "stays flat")
avg = sum(vals) / len(vals)
return (
f"**{name}** {trend} from **{v0:.1f}%** in {first} to **{vN:.1f}%** in "
f"{last}{scen_suffix} โ average of **{avg:.1f}%** across the {len(vals)} months.\n\n"
f"- **Highest:** {max(vals):.1f}% ยท **Lowest:** {min(vals):.1f}%\n"
f"- **Net change:** {_format_pct(vN - v0)}\n"
)
trend = "rises" if vN > v0 else ("falls" if vN < v0 else "stays flat")
total = sum(vals)
return (
f"**{name}** {trend} from **{_format_kwacha(v0)}** in {first} to "
f"**{_format_kwacha(vN)}** in {last}{scen_suffix}{_delta_pct(v0, vN)}.\n\n"
f"- **Total across the period:** {_format_kwacha(total)}\n"
f"- **Average per month:** {_format_kwacha(total / len(vals))}\n"
)
# Two-series โ typical revenue vs expenses, profit vs margin โโโโโโโโโ
if len(datasets) == 2:
a, b = datasets[0], datasets[1]
name_a = (a.get("name") or "Series A").strip()
name_b = (b.get("name") or "Series B").strip()
va, vb = a.get("values") or [], b.get("values") or []
if not va or not vb:
return ""
a_pct = "%" in name_a or "margin" in name_a.lower()
b_pct = "%" in name_b or "margin" in name_b.lower()
fmt_a = (lambda v: f"{v:.1f}%") if a_pct else _format_kwacha
fmt_b = (lambda v: f"{v:.1f}%") if b_pct else _format_kwacha
lines = [
f"Here's how **{name_a}** and **{name_b}** move from {first} to "
f"{last}{scen_suffix}:",
"",
f"- **{name_a}:** {fmt_a(va[0])} โ {fmt_a(va[-1])}"
+ (f" {_delta_pct(va[0], va[-1])}" if not a_pct else f" ({(va[-1] - va[0]):+.1f} pts)"),
f"- **{name_b}:** {fmt_b(vb[0])} โ {fmt_b(vb[-1])}"
+ (f" {_delta_pct(vb[0], vb[-1])}" if not b_pct else f" ({(vb[-1] - vb[0]):+.1f} pts)"),
]
# If looks like Revenue vs Expenses, add net (gap)
if not a_pct and not b_pct and len(va) == len(vb):
gap_first = va[0] - vb[0]
gap_last = va[-1] - vb[-1]
lines.append(
f"- **Gap ({name_a} โ {name_b}):** "
f"{_format_kwacha(gap_first)} โ {_format_kwacha(gap_last)}"
)
return "\n".join(lines)
# Three-series fallback (e.g. Revenue + Expenses + Profit) โโโโโโโโโโโ
try:
names = [(d.get("name") or "?").strip() for d in datasets]
firsts = [d.get("values", [None])[0] for d in datasets]
lasts = [d.get("values", [None])[-1] for d in datasets]
lines = [
f"Here's the breakdown from {first} to {last}{scen_suffix}:",
"",
]
for n, v0, vN in zip(names, firsts, lasts):
if v0 is None or vN is None:
continue
is_pct = "%" in n or "margin" in n.lower()
fmt = (lambda v: f"{v:.1f}%") if is_pct else _format_kwacha
lines.append(f"- **{n}:** {fmt(v0)} โ {fmt(vN)}")
return "\n".join(lines)
except Exception:
return f"{title or 'Forecast view'}: {', '.join(names)} from {first} to {last}{scen_suffix}."
except Exception:
return ""
def _forecast_series_is_percentage_axis(name: str) -> bool:
"""Names that denote margin / % โ must use a separate axis from Kwacha amounts."""
if not name:
return False
if "%" in name:
return True
low = name.lower()
return "margin" in low
def build_plotly_chart(chart_data: dict) -> go.Figure:
"""Build a dark-themed Plotly chart from AI-generated chart JSON data."""
fig = go.Figure()
chart_type = chart_data.get("type", "line")
labels = chart_data["data"]["labels"]
datasets = chart_data["data"]["datasets"]
pct_flags = [_forecast_series_is_percentage_axis(d.get("name", "")) for d in datasets]
dual_axis = bool(datasets) and any(pct_flags) and not all(pct_flags)
for dataset in datasets:
name = dataset["name"]
values = dataset["values"]
per_bar_colors = dataset.get("colors")
raw_color = dataset.get("color", "#1D9E75")
if per_bar_colors is None and isinstance(raw_color, list):
per_bar_colors = raw_color
single_color = "#1D9E75"
else:
single_color = raw_color if isinstance(raw_color, str) else "#1D9E75"
is_pct_series = _forecast_series_is_percentage_axis(name)
yaxis_ref = "y2" if dual_axis and is_pct_series else "y"
if chart_type == "line":
hover = (
f"%{{x}}: %{{y:.1f}}%{name}"
if is_pct_series
else f"%{{x}}: K%{{y:,.0f}}{name}"
)
fig.add_trace(go.Scatter(
x=labels,
y=values,
name=name,
yaxis=yaxis_ref,
line=dict(color=single_color, width=3),
mode="lines+markers",
marker=dict(size=9, color=single_color,
line=dict(width=1.5, color="rgba(255,255,255,0.35)")),
hovertemplate=hover,
))
elif chart_type == "bar":
is_margin = is_pct_series
hover = (
f"%{{x}}: %{{y:.1f}}%{name}"
if is_margin
else f"%{{x}}: K%{{y:,.0f}}{name}"
)
n_bars = len(values)
if per_bar_colors:
if len(per_bar_colors) < n_bars:
last = per_bar_colors[-1] if per_bar_colors else single_color
bar_colors = list(per_bar_colors) + [last] * (n_bars - len(per_bar_colors))
else:
bar_colors = list(per_bar_colors)[:n_bars]
else:
bar_colors = [single_color] * n_bars
fig.add_trace(go.Bar(
x=labels,
y=values,
name=name,
yaxis=yaxis_ref,
marker=dict(color=bar_colors, line=dict(width=0)),
hovertemplate=hover,
text=[f"{v:.1f}%" if is_margin else f"K{v:,.0f}" for v in values],
textposition="outside",
textfont=dict(size=11, color="#E2E8F0"),
))
only_pct_single_axis = bool(datasets) and all(pct_flags)
fig.update_layout(
title=dict(
text=chart_data.get("title", "Financial Chart"),
font=dict(size=17, color="#F1F5F9", family="Segoe UI, Arial, sans-serif"),
x=0.5,
xanchor="center",
),
paper_bgcolor="rgba(0,0,0,0)",
plot_bgcolor="rgba(13,31,60,0.88)",
font=dict(color="#E2E8F0", size=14, family="Segoe UI, Arial, sans-serif"),
legend=dict(
bgcolor="rgba(15,23,42,0.78)",
bordercolor="rgba(148,163,184,0.22)",
borderwidth=1,
font=dict(size=13),
orientation="h",
y=1.09,
x=0.5,
xanchor="center",
),
xaxis=dict(
gridcolor="rgba(148,163,184,0.1)",
color="#E2E8F0",
tickfont=dict(size=13),
showline=True,
linecolor="rgba(148,163,184,0.25)",
),
margin=dict(l=56, r=56 if dual_axis else 28, t=76, b=44),
height=int(chart_data.get("height", 455)),
hovermode="x unified",
bargap=0.28,
)
if dual_axis:
fig.update_layout(
yaxis=dict(
title=dict(text="Amount (Kwacha)", font=dict(size=13)),
gridcolor="rgba(148,163,184,0.12)",
color="#93C5FD",
tickprefix="K",
ticksuffix="",
zeroline=True,
zerolinecolor="rgba(148,163,184,0.35)",
zerolinewidth=1,
tickfont=dict(size=13),
),
yaxis2=dict(
title=dict(text="Margin / %", font=dict(size=13)),
overlaying="y",
side="right",
gridcolor="rgba(148,163,184,0.06)",
color="#FCA5A5",
tickprefix="",
ticksuffix="%",
zeroline=False,
showgrid=False,
tickfont=dict(size=13),
),
)
else:
fig.update_layout(
yaxis=dict(
title=dict(
text="Margin (%)" if only_pct_single_axis else "Amount (Kwacha)",
font=dict(size=13),
),
gridcolor="rgba(148,163,184,0.12)",
color="#CBD5E1",
tickprefix="" if only_pct_single_axis else "K",
ticksuffix="%" if only_pct_single_axis else "",
zeroline=True,
zerolinecolor="rgba(148,163,184,0.3)",
zerolinewidth=1,
tickfont=dict(size=13),
),
)
return fig
def render_cash_flow_forecast_page() -> None:
render_page_header(
"Cash Flow Forecast",
"Project the next 6 months using your current revenue and expenses.",
"6 months",
)
# โโ Resolve baseline (single source of truth) โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ
baseline = st.session_state.get("baseline") or get_baseline(st.session_state)
st.markdown(
'',
unsafe_allow_html=True,
)
if not baseline:
st.markdown(
f''
f'โ No financial data found.
'
f'Go to Chat Advisor and run an analysis first, or enter your numbers in the Business Context form.'
f'
',
unsafe_allow_html=True,
)
if st.button("โ Go to Chat Advisor"):
st.session_state.active_page = "Chat Advisor"
st.rerun()
return
# Data source banner
months_note = (
f" ยท Average of {baseline['months_of_data']} months of actual data"
if baseline["months_of_data"] > 1 else ""
)
st.markdown(
f''
f'๐ Forecast based on: {baseline["source"]}'
f' ยท Starting monthly revenue: K{baseline["monthly_revenue"]:,.2f}'
f' ยท Expenses: K{baseline["monthly_expenses"]:,.2f}'
f'{months_note}
',
unsafe_allow_html=True,
)
# โโ Scenario selector โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ
if "forecast_scenario" not in st.session_state:
st.session_state["forecast_scenario"] = "base_case"
s_col1, s_col2, s_col3 = st.columns(3, gap="small")
for col, key in zip([s_col1, s_col2, s_col3], ["pessimistic", "base_case", "optimistic"]):
sc = FE_SCENARIOS[key]
active = st.session_state["forecast_scenario"] == key
btn_label = f"{sc['icon']} **{sc['label']}**\n{sc['description']}"
if col.button(btn_label, key=f"sc_{key}", use_container_width=True,
type="primary" if active else "secondary"):
st.session_state["forecast_scenario"] = key
for k in list(st.session_state.keys()):
if k.startswith("forecast_summary_"):
del st.session_state[k]
st.rerun()
scenario_key = st.session_state["forecast_scenario"]
forecast = calculate_forecast(baseline, scenario_key, months=6)
s = forecast["summary"]
rows = forecast["months"]
# โโ Single-column dashboard (full width). Conversation lives below in a
# centered, narrower column so chat doesn't sprawl across the whole page.
st.markdown('', unsafe_allow_html=True)
# KPI strip โ full width
k1, k2, k3, k4 = st.columns(4, gap="small")
profit_now = baseline["monthly_profit"]
rev_now = baseline["monthly_revenue"]
exp_now = baseline["monthly_expenses"]
k1.metric(
"Month 6 Revenue", f"K{s['final_revenue']:,.0f}",
f"{(s['final_revenue']-rev_now)/max(rev_now,1)*100:+.0f}% vs now",
)
k2.metric(
"Month 6 Expenses", f"K{s['final_expenses']:,.0f}",
f"{(s['final_expenses']-exp_now)/max(exp_now,1)*100:+.0f}% vs now",
)
profit_delta = s["final_profit"] - profit_now
k3.metric(
"Month 6 Profit", f"K{s['final_profit']:,.0f}",
f"K{profit_delta:+,.0f} vs now",
delta_color="normal" if profit_delta >= 0 else "inverse",
)
k4.metric(
"Profitable Months", f"{s['profitable_months']}/6",
"All positive" if s["all_profitable"] else f"Risk from {s['first_loss_month']}",
delta_color="off" if s["all_profitable"] else "inverse",
)
# โโ Chart + AI summary side-by-side. Both end at roughly the same height
# so there is no orphaned empty space when conversation grows below.
col_chart, col_ai = st.columns([1.75, 1], gap="large")
with col_chart:
# Area + Bar chart
month_labels = [r["month_name"] for r in rows]
rev_vals = [r["revenue"] for r in rows]
exp_vals = [r["expenses"] for r in rows]
profit_vals = [r["profit"] for r in rows]
bar_colors = [THEME["emerald"] if p >= 0 else THEME["red"] for p in profit_vals]
fig = go.Figure()
fig.add_trace(go.Scatter(
x=month_labels, y=rev_vals, name="Revenue",
mode="lines+markers", line=dict(color=THEME["emerald"], width=2.5),
fill="tozeroy", fillcolor="rgba(16,185,129,0.07)", marker=dict(size=6),
))
fig.add_trace(go.Scatter(
x=month_labels, y=exp_vals, name="Expenses",
mode="lines+markers", line=dict(color=THEME["orange"], width=2.5, dash="dot"),
fill="tozeroy", fillcolor="rgba(249,115,22,0.05)", marker=dict(size=6),
))
fig.add_trace(go.Bar(
x=month_labels, y=profit_vals, name="Profit",
marker_color=bar_colors, opacity=0.75, yaxis="y2",
))
fig.add_hline(
y=0, line_dash="dash", line_color="rgba(148,163,184,0.4)",
annotation_text="Break-even", annotation_position="bottom right",
annotation_font=dict(color="rgba(148,163,184,0.6)", size=10),
)
fig.update_layout(
height=480,
margin=dict(l=8, r=12, t=110, b=12),
paper_bgcolor="rgba(0,0,0,0)", plot_bgcolor="rgba(15,23,42,0.35)",
font=dict(color="#E2E8F0", size=14, family="Segoe UI, Arial, sans-serif"),
title=dict(
text="
6-month outlook โ revenue, expenses & profit",
font=dict(size=17, color="#F8FAFC"),
x=0.005,
xanchor="left",
y=0.97,
yanchor="top",
),
xaxis=dict(showgrid=False, tickfont=dict(size=13), color="#94A3B8"),
yaxis=dict(
title=dict(text="Revenue & expenses (K)", font=dict(size=13)),
gridcolor="rgba(148,163,184,0.12)",
zeroline=False,
tickfont=dict(size=13),
tickprefix="K",
),
yaxis2=dict(
title=dict(text="Profit (K)", font=dict(size=13)),
overlaying="y",
side="right",
showgrid=False,
zeroline=True,
zerolinecolor="rgba(148,163,184,0.35)",
tickfont=dict(size=13),
tickprefix="K",
),
legend=dict(
orientation="h",
y=1.07,
yanchor="bottom",
x=0,
xanchor="left",
font=dict(size=12.5),
bgcolor="rgba(15,23,42,0)",
),
bargap=0.32, hovermode="x unified",
)
st.plotly_chart(fig, use_container_width=True)
# Insight strip
if s["first_loss_month"]:
st.markdown(
f'
'
f'โ ๏ธ Expenses may exceed revenue from {s["first_loss_month"]} under this scenario.
',
unsafe_allow_html=True,
)
else:
st.markdown(
f'
'
f'โ
All 6 months profitable. Month 6 margin: {s["final_margin"]:.1f}%.
',
unsafe_allow_html=True,
)
# Monthly breakdown table
with st.expander("๐ Monthly Breakdown Table", expanded=False):
def _tr(r: dict, prev_profit: float | None) -> str:
pc = THEME["emerald"] if r["profit"] >= 0 else THEME["red"]
if prev_profit is not None:
diff = r["profit"] - prev_profit
trend_html = (
f'
โK{diff:,.0f}'
if diff >= 0
else f'
โK{abs(diff):,.0f}'
)
else:
trend_html = ""
warn = " โ ๏ธ" if not r["profitable"] else ""
return (
f'
'
f'| {r["month_name"]} | '
f'K{r["revenue"]:,.0f} | '
f'K{r["expenses"]:,.0f} | '
f'K{r["profit"]:,.0f}{warn}{trend_html} | '
f'{r["margin"]:.1f}% | '
f'
'
)
table_html = "".join(
_tr(r, rows[i-1]["profit"] if i > 0 else None)
for i, r in enumerate(rows)
)
st.markdown(
'
'
''
'| Month | '
'Revenue | '
'Expenses | '
'Profit | '
'Margin | '
f'
{table_html}
',
unsafe_allow_html=True,
)
# โโ AI analyst summary (right side of dashboard) โโโโโโโโโโโโโโโโโโโโโโโโโโโ
with col_ai:
st.markdown(
'
'
'
'
"AI analyst
"
'
Forecast snapshot
'
'
'
"A plain-English read of the chart on the left.
"
"
",
unsafe_allow_html=True,
)
cache_key = f"forecast_summary_{scenario_key}"
if cache_key not in st.session_state:
with st.spinner("๐ฎ Writing AI forecast narrativeโฆ"):
st.session_state[cache_key] = _generate_forecast_ai_summary(
baseline, forecast, scenario_key
)
st.markdown(
'
' + html.escape(
st.session_state[cache_key]
).replace("\n", "
")
+ "
",
unsafe_allow_html=True,
)
sugg_pills = [
("๐ Profit trend", "Show profit trend"),
("๐ Revenue vs expenses", "Revenue vs expenses"),
("โ ๏ธ Riskiest months", "Riskiest months"),
("๐ฐ Expense breakdown", "Expense breakdown"),
]
st.caption("Quick prompts ยท tap to ask")
for idx, (label, prompt_text) in enumerate(sugg_pills):
if st.button(label, key=f"fpill_{idx}", use_container_width=True):
st.session_state["forecast_question"] = prompt_text
st.session_state["forecast_pin_to_question"] = True
st.rerun()
# End dashboard wrapper
st.markdown("
'
'
'
'๐ฌ Chat with the analyst'
'Ask anything โ answers use the numbers from this scenario.'
"
",
unsafe_allow_html=True,
)
if not st.session_state["forecast_chat"]:
st.markdown(
'
No questions yet โ try a quick prompt above '
'or type below.
',
unsafe_allow_html=True,
)
for _msg_idx, msg in enumerate(st.session_state["forecast_chat"]):
if msg["role"] == "user":
escaped = (
msg["content"]
.replace("&", "&")
.replace("<", "<")
.replace(">", ">")
)
st.markdown(
f'
',
unsafe_allow_html=True,
)
else:
with st.chat_message("assistant", avatar="๐"):
body = sanitize_forecast_answer_text(msg["content"])
st.markdown(body)
if msg.get("plotly_chart"):
st.plotly_chart(
build_plotly_chart(msg["plotly_chart"]),
use_container_width=True,
key=f"forecast_chat_plotly_{_msg_idx}",
)
elif msg.get("chart"):
render_followup_chart(msg["chart"])
st.markdown("
'
'
Analysis required
'
'
Run Chat Advisor first
'
'
The Scenario Planner needs your actual revenue and expenses before it can calculate anything useful.
'
'
'
f'๐ Current baseline: {baseline["source"]}'
f' ยท Revenue K{baseline["monthly_revenue"]:,.2f}'
f' ยท Expenses K{baseline["monthly_expenses"]:,.2f}'
f' ยท Profit K{baseline["monthly_profit"]:,.2f}'
f' ยท Margin {baseline["profit_margin"]:.1f}%'
f'
',
unsafe_allow_html=True,
)
render_tool_tip("How to use", "Move the sliders to model a what-if scenario. All metrics recalculate instantly in Python โ no AI required.")
# โโ Sliders โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ
ctrl1, ctrl2, ctrl3 = st.columns(3, gap="large")
with ctrl1:
revenue_delta = st.slider("Revenue change", -50, 75, 0, 5, format="%d%%",
help="What if your sales grew or fell by this percentage?")
with ctrl2:
expense_delta = st.slider("Expense change", -50, 50, 0, 5, format="%d%%",
help="What if your total costs changed by this percentage?")
with ctrl3:
extra_debt = st.number_input(
"Extra monthly debt payment (K)", min_value=0.0, value=0.0, step=100.0,
help="Add a hypothetical loan repayment to see the impact.",
)
# โโ Python does ALL the math โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ
new_rev = baseline["monthly_revenue"] * (1 + revenue_delta / 100)
new_exp = baseline["monthly_expenses"] * (1 + expense_delta / 100) + extra_debt
scenario_baseline = {
**baseline,
"monthly_revenue": round(new_rev, 2),
"monthly_expenses": round(new_exp, 2),
"monthly_profit": round(new_rev - new_exp, 2),
"profit_margin": round((new_rev - new_exp) / new_rev * 100, 1) if new_rev > 0 else 0.0,
}
new_metrics = compute_metrics(scenario_baseline)
delta_profit = new_metrics["monthly_profit"] - base_metrics["monthly_profit"]
delta_health = new_metrics["health_score"] - base_metrics["health_score"]
delta_margin = new_metrics["profit_margin_pct"] - base_metrics["profit_margin_pct"]
# โโ KPI cards โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ
cols = st.columns(4, gap="small")
cols[0].metric("New Revenue", f"K{new_rev:,.0f}", f"{revenue_delta:+.0f}%")
cols[1].metric("New Expenses", f"K{new_exp:,.0f}", f"{expense_delta:+.0f}%")
cols[2].metric(
"New Profit", f"K{new_metrics['monthly_profit']:,.0f}",
f"K{delta_profit:+,.0f}",
delta_color="normal" if delta_profit >= 0 else "inverse",
)
cols[3].metric(
"Health Score", f"{new_metrics['health_score']}/100",
f"{delta_health:+.0f} pts",
delta_color="normal" if delta_health >= 0 else "inverse",
)
# Margin and loan label change
loan_label, loan_color, loan_icon = get_loan_label(new_metrics["profit_margin_pct"])
m_col1, m_col2 = st.columns(2, gap="large")
m_col1.metric(
"New Margin", f"{new_metrics['profit_margin_pct']:.1f}%",
f"{delta_margin:+.1f}pp vs current",
delta_color="normal" if delta_margin >= 0 else "inverse",
)
m_col2.metric(
"Loan Readiness", f"{loan_icon} {loan_label}",
f"Safe borrowing: K{new_metrics['safe_loan_amount']:,.0f}",
delta_color="off",
)
# Verdict
if delta_health > 10:
st.success(f"โ
This scenario improves your business health by {delta_health} points.")
elif delta_health < -10:
st.error(f"โ This scenario reduces your business health by {abs(delta_health)} points.")
elif new_metrics["monthly_profit"] <= 0:
st.error("โ This scenario results in a loss. Expenses exceed revenue.")
else:
st.info("โน๏ธ This scenario has minimal impact on overall business health.")
# Safe loan formula display
st.caption(
f"Safe borrowing amount under this scenario: K{new_metrics['safe_loan_amount']:,.2f} "
f"({new_metrics['safe_loan_formula']}) โ a conservative SME lending rule of thumb."
)
# โโ Send to Chat button โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ
st.divider()
if st.button("๐ฌ Ask Duka AI to explain this scenario", type="primary"):
scenario_desc = (
f"I modelled a scenario: revenue {revenue_delta:+.0f}%, "
f"expenses {expense_delta:+.0f}%"
+ (f", extra debt payment K{extra_debt:,.0f}/month" if extra_debt > 0 else "")
+ f". New profit would be K{new_metrics['monthly_profit']:,.0f} "
f"(was K{base_metrics['monthly_profit']:,.0f}). "
f"Health score changes from {base_metrics['health_score']} to {new_metrics['health_score']}. "
f"Margin changes from {base_metrics['profit_margin_pct']:.1f}% to {new_metrics['profit_margin_pct']:.1f}%. "
"What should I focus on given these numbers?"
)
st.session_state["pending_chat_question"] = scenario_desc
st.session_state.active_page = "Chat Advisor"
st.rerun()
def render_loan_calculator_page() -> None:
render_page_header(
"Loan Calculator",
"Estimate monthly repayment and compare it against current profit.",
"Repayment",
)
report = require_report()
if not report:
return
loan = report["loan"]
cashflow = report["cashflow"]
render_tool_tip("Borrowing guardrail", "A repayment above roughly 20-30% of current profit is usually a warning sign for a small business.")
control_cols = st.columns(3, gap="large")
with control_cols[0]:
amount = st.number_input("Loan amount (K)", min_value=0.0, value=float(loan.get("suggested_loan_amount", 0.0)), step=500.0)
with control_cols[1]:
annual_rate = st.slider("Annual interest rate", 0.0, 60.0, 28.0, 0.5, format="%.1f%%")
with control_cols[2]:
months = st.slider("Repayment period (months)", 1, 60, 12)
monthly_rate = annual_rate / 100 / 12
payment = amount / months if monthly_rate == 0 else amount * monthly_rate / (1 - (1 + monthly_rate) ** -months)
monthly_profit = max(float(cashflow.get("profit", 0.0)), 0.0)
burden = (payment / monthly_profit * 100) if monthly_profit else 0
cols = st.columns(3)
cols[0].metric("Monthly repayment", format_currency(payment))
cols[1].metric("Total repayment", format_currency(payment * months))
cols[2].metric("Profit used", f"{burden:.1f}%")
if monthly_profit <= 0:
st.error("Current profit is not positive, so new borrowing is not advisable from these numbers.")
elif burden > 30:
st.warning("This repayment would use a high share of current profit.")
else:
st.success("This repayment is within a more manageable range based on current profit.")
render_loan_readiness_card(report)
def get_expense_breakdown(report: dict[str, Any]) -> dict[str, float]:
parsed = report.get("parsed_data", {})
breakdown = {str(k).replace("_", " ").title(): float(v or 0) for k, v in (parsed.get("expenses_breakdown") or {}).items() if float(v or 0) > 0}
if breakdown:
return breakdown
doc_breakdown = report.get("document_analysis", {}).get("expenses_breakdown") or {}
return {str(k).replace("_", " ").title(): float(v or 0) for k, v in doc_breakdown.items() if float(v or 0) > 0}
def _expense_identity_reply() -> str:
return (
"I'm **Duka AI's Expense Analyst** โ I focus only on **where your Kwacha is going**. "
"I work from the verified expense breakdown on this page (no guesses): I can rank your top "
"categories, show what % of revenue each one eats, and suggest **specific** cost cuts.\n\n"
"Try: *Which category is hurting me most?* ยท *How can I cut transport costs?* ยท "
"*Where should I trim 10% of expenses?*"
)
def _build_expense_chat_context(
report: dict[str, Any], breakdown: dict[str, float], frame: "pd.DataFrame"
) -> str:
cf = report.get("cashflow") or {}
revenue = float(cf.get("revenue", 0) or 0)
total_exp = float(frame["Amount"].sum())
rows = frame.sort_values("Amount", ascending=False).to_dict("records")
bp = report.get("business_profile") or {}
bt = (bp.get("business_type") or "").strip() or "SME"
loc = (bp.get("location") or "").strip() or "Zambia"
lines: list[str] = []
lines.append("VERIFIED EXPENSE BREAKDOWN (Python-computed, do NOT invent figures):\n")
lines.append(
f"- Business: {bt} ยท Location: {loc}\n"
f"- Monthly revenue: K{revenue:,.0f}\n"
f"- Total monthly expenses: K{total_exp:,.0f}"
+ (f" ({total_exp / revenue * 100:.1f}% of revenue)\n" if revenue > 0 else "\n")
)
lines.append("- Categories (largest โ smallest):\n")
for r in rows:
cat = str(r["Category"])
amt = float(r["Amount"])
pct_rev = (amt / revenue * 100) if revenue > 0 else 0.0
pct_exp = (amt / total_exp * 100) if total_exp > 0 else 0.0
lines.append(
f" ยท {cat}: K{amt:,.0f} ({pct_exp:.1f}% of expenses, {pct_rev:.1f}% of revenue)\n"
)
return "".join(lines)
def _classify_expense_chat_input(question: str) -> str | None:
low = (question or "").lower().strip()
if not low:
return "empty"
if any(p in low for p in ("who are you", "what are you", "what do you do", "introduce yourself")):
return "identity"
if low in ("thanks", "thank you", "ty", "cheers", "ta"):
return "thanks"
return None
def _handle_expense_question(
question: str,
report: dict,
breakdown: dict[str, float],
frame: "pd.DataFrame",
chat_history: list[dict],
) -> str:
from agents import request_llm_chat
canned = _classify_expense_chat_input(question)
if canned == "identity":
return _expense_identity_reply()
if canned == "thanks":
return "Glad it helped โ ask anything else about your expense breakdown."
if canned == "empty":
return "Type a question about your expense categories or how to cut them."
ctx = _build_expense_chat_context(report, breakdown, frame)
system = (
"You are **Duka AI's Expense Analyst** for **Zambian SMEs**. "
"You ONLY answer using the VERIFIED EXPENSE BREAKDOWN below โ never invent categories or amounts. "
"When you reference a category, **always** quote its exact K amount and its % of revenue or % of expenses. "
"Be specific with cost-cutting advice (which line, by how much in K, and why it's realistic for that business type/location). "
"Use **K** for Kwacha. Reply in 3โ6 short sentences or short bullets โ no fluff.\n\n"
f"{ctx}"
)
messages = [
{"role": "system", "content": system},
*[{"role": m["role"], "content": m["content"]} for m in chat_history[-8:]],
{"role": "user", "content": question},
]
result = request_llm_chat(messages, temperature=0.2, max_tokens=500)
return result or "I couldn't generate a response right now. Please try again."
def render_expense_analyzer_page() -> None:
render_page_header(
"Expense Analyzer",
"Break down where money is going and spot the biggest cost pressure.",
"Donut + AI agent",
)
report = require_report()
if not report:
return
st.markdown(
'',
unsafe_allow_html=True,
)
cashflow = report["cashflow"]
breakdown = get_expense_breakdown(report)
if not breakdown:
breakdown = {"Total expenses": float(cashflow["expenses"])}
st.info("Detailed expense categories were not provided, so this view shows total expenses only.")
frame = pd.DataFrame(
[{"Category": k, "Amount": v} for k, v in breakdown.items()]
).sort_values("Amount", ascending=False)
top_row = frame.iloc[0]
# Two-column layout: snapshot/chart left, AI agent chat right
col_chart, col_chat = st.columns([3, 2], gap="large")
with col_chart:
metric_cols = st.columns(3)
metric_cols[0].metric("Total expenses", format_currency(float(frame["Amount"].sum())))
metric_cols[1].metric("Largest category", str(top_row["Category"]))
metric_cols[2].metric("Largest amount", format_currency(float(top_row["Amount"])))
st.dataframe(frame, use_container_width=True, hide_index=True)
fig = go.Figure(data=[go.Pie(labels=frame["Category"], values=frame["Amount"], hole=0.58)])
fig.update_layout(
height=380,
margin=dict(l=8, r=8, t=16, b=8),
paper_bgcolor="rgba(0,0,0,0)",
font=dict(color="#E2E8F0", family="Segoe UI, Arial, sans-serif"),
legend=dict(orientation="h", y=-0.05),
)
st.plotly_chart(fig, use_container_width=True)
with col_chat:
st.markdown(
'',
unsafe_allow_html=True,
)
st.markdown(
''
'
'
'๐ Ask the Expense Analyst'
''
"Grounded in your verified categories โ no invented numbers."
""
"
",
unsafe_allow_html=True,
)
sugg_pills = [
"๐ฅ Which category hurts me most?",
"โ๏ธ Where can I cut 10%?",
"๐ How do I reduce transport?",
"๐ฅ Are wages too high?",
"๐ก Which line saves me the most?",
"๐ What % of revenue goes to expenses?",
]
p1, p2 = st.columns(2)
for idx, pill in enumerate(sugg_pills):
col = p1 if idx % 2 == 0 else p2
if col.button(pill, key=f"epill_{idx}", use_container_width=True):
st.session_state["expense_question"] = pill
st.markdown("
", unsafe_allow_html=True)
if "expense_chat" not in st.session_state:
st.session_state["expense_chat"] = []
if not st.session_state["expense_chat"]:
st.markdown(
'
No messages yet โ tap a suggestion '
"or type below.
",
unsafe_allow_html=True,
)
for msg in st.session_state["expense_chat"]:
if msg["role"] == "user":
escaped = (
msg["content"]
.replace("&", "&")
.replace("<", "<")
.replace(">", ">")
)
st.markdown(
f'
',
unsafe_allow_html=True,
)
else:
with st.chat_message("assistant", avatar="๐"):
st.markdown(msg["content"])
st.markdown("
")
return Paragraph(t, style)
def build_full_report_pdf(full: dict) -> bytes:
"""Build a styled PDF from generate_full_report(); falls back if ReportLab is missing or fails."""
try:
return _render_reportlab_pdf(full)
except Exception:
return _build_plaintext_pdf_from_full(full)
def _render_reportlab_pdf(full: dict) -> bytes:
"""Premium ReportLab layout (requires `reportlab` package)."""
from io import BytesIO
from reportlab.lib.pagesizes import A4
from reportlab.lib.styles import getSampleStyleSheet, ParagraphStyle
from reportlab.lib.units import cm
from reportlab.lib import colors
from reportlab.platypus import (
SimpleDocTemplate,
Paragraph,
Spacer,
Table,
TableStyle,
HRFlowable,
KeepTogether,
)
buf = BytesIO()
doc = SimpleDocTemplate(
buf,
pagesize=A4,
rightMargin=1.8 * cm,
leftMargin=1.8 * cm,
topMargin=1.6 * cm,
bottomMargin=1.8 * cm,
)
base = getSampleStyleSheet()
C_DARK = colors.HexColor("#0F172A")
C_BLUE = colors.HexColor("#2563EB")
C_SLATE = colors.HexColor("#1e293b")
C_MID = colors.HexColor("#64748B")
C_LITE = colors.HexColor("#F1F5F9")
C_CARD = colors.HexColor("#f8fafc")
C_GRID = colors.HexColor("#CBD5E1")
C_GRN = colors.HexColor("#059669")
C_RED = colors.HexColor("#DC2626")
C_AMB = colors.HexColor("#D97706")
def _style(name, **kw):
return ParagraphStyle(name, parent=base["Normal"], **kw)
ST_SUB = _style(
"RP_SUB",
fontSize=9.5,
textColor=colors.HexColor("#cbd5e1"),
leading=13,
)
NM = _style("RP_NM", fontSize=10, textColor=C_DARK, leading=15)
NM_SM = _style("RP_NM_SM", fontSize=9, textColor=C_DARK, leading=13)
SM = _style("RP_SM", fontSize=9, textColor=C_MID, leading=13)
H3 = _style(
"RP_H3",
fontSize=10,
textColor=C_MID,
fontName="Helvetica-Bold",
spaceBefore=6,
spaceAfter=2,
)
FT = _style("RP_FT", fontSize=8, textColor=C_MID, alignment=1)
TITLE = _style(
"RP_TITLE",
fontSize=20,
textColor=colors.white,
fontName="Helvetica-Bold",
leading=24,
spaceAfter=4,
)
snap = full.get("snapshot", {})
h_score = int(snap.get("health_score", 0) or 0)
h_color = C_GRN if h_score >= 70 else (C_AMB if h_score >= 40 else C_RED)
loan = full.get("loan", {})
l_score = int(loan.get("score", 0) or 0)
l_color = C_GRN if l_score >= 70 else (C_AMB if l_score >= 40 else C_RED)
def _hr(thickness=0.8, c=C_GRID):
return HRFlowable(width="100%", thickness=thickness, color=c, spaceAfter=8, spaceBefore=8)
def _tbl(data, col_widths, style_cmds):
t = Table(data, colWidths=col_widths)
t.setStyle(
TableStyle(
[
("FONTNAME", (0, 0), (-1, -1), "Helvetica"),
("FONTSIZE", (0, 0), (-1, -1), 10),
(
"ROWBACKGROUNDS",
(0, 0),
(-1, -1),
[colors.white, C_CARD],
),
("GRID", (0, 0), (-1, -1), 0.35, C_GRID),
("TOPPADDING", (0, 0), (-1, -1), 6),
("BOTTOMPADDING", (0, 0), (-1, -1), 6),
("LEFTPADDING", (0, 0), (-1, -1), 8),
]
+ style_cmds
)
)
return t
def _section_bar(label: str):
lab = _style(
"RP_SEC",
fontSize=9,
textColor=colors.white,
fontName="Helvetica-Bold",
leading=11,
letterSpacing=1.2,
)
tb = Table([[Paragraph(_pdf_escape(label.upper()), lab)]], colWidths=[17.4 * cm])
tb.setStyle(
TableStyle(
[
("BACKGROUND", (0, 0), (-1, -1), C_BLUE),
("TOPPADDING", (0, 0), (-1, -1), 8),
("BOTTOMPADDING", (0, 0), (-1, -1), 8),
("LEFTPADDING", (0, 0), (-1, -1), 10),
]
)
)
return tb
story = []
# โโ Hero header โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ
hero = Table(
[
[Paragraph(_pdf_escape("Duka AI Financial Health Report"), TITLE)],
[
_pdf_para(
f"{full.get('business_name', 'Your Business')} | "
f"{full.get('location', 'Zambia')} | "
f"Generated {full.get('generated_at', '')}",
ST_SUB,
)
],
],
colWidths=[17.4 * cm],
)
hero.setStyle(
TableStyle(
[
("BACKGROUND", (0, 0), (-1, -1), C_SLATE),
("TOPPADDING", (0, 0), (-1, -1), 16),
("BOTTOMPADDING", (0, 0), (-1, -1), 14),
("LEFTPADDING", (0, 0), (-1, -1), 14),
("RIGHTPADDING", (0, 0), (-1, -1), 14),
("LINEABOVE", (0, 0), (-1, 0), 3, C_BLUE),
]
)
)
story.append(hero)
_snap_profit = float(snap.get("profit", 0) or 0)
_pr_col = C_RED if _snap_profit < 0 else C_GRN
_rev = float(snap.get("revenue", 0) or 0)
_exp = float(snap.get("expenses", 0) or 0)
kpi = Table(
[
[
"MONTHLY REVENUE",
"MONTHLY EXPENSES",
"NET " + ("LOSS" if _snap_profit < 0 else "PROFIT"),
],
[
f"K{_rev:,.0f}",
f"K{_exp:,.0f}",
f"K{abs(_snap_profit):,.0f}",
],
],
colWidths=[5.8 * cm, 5.8 * cm, 5.8 * cm],
)
kpi.setStyle(
TableStyle(
[
("BACKGROUND", (0, 0), (-1, -1), C_LITE),
("BOX", (0, 0), (-1, -1), 0.75, C_GRID),
("FONTNAME", (0, 0), (-1, 0), "Helvetica-Bold"),
("FONTSIZE", (0, 0), (-1, 0), 7.5),
("TEXTCOLOR", (0, 0), (-1, 0), C_MID),
("FONTNAME", (0, 1), (-1, 1), "Helvetica-Bold"),
("FONTSIZE", (0, 1), (-1, 1), 14),
("TEXTCOLOR", (0, 1), (1, 1), C_DARK),
("TEXTCOLOR", (2, 1), (2, 1), _pr_col),
("ALIGN", (0, 0), (-1, -1), "CENTER"),
("TOPPADDING", (0, 0), (-1, -1), 10),
("BOTTOMPADDING", (0, 0), (-1, -1), 12),
]
)
)
story.append(kpi)
story.append(Spacer(1, 0.35 * cm))
# โโ Executive Summary โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ
story.append(_section_bar("Executive summary"))
story.append(Spacer(1, 0.15 * cm))
story.append(_pdf_para(full.get("executive_summary", ""), NM))
story.append(Spacer(1, 0.3 * cm))
# โโ Financial Snapshot โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ
story.append(_section_bar("Financial snapshot"))
story.append(Spacer(1, 0.15 * cm))
_snap_p_label = "Net Loss" if _snap_profit < 0 else "Profit"
_snap_p_value = (
f"K{abs(_snap_profit):,.0f} ({float(snap.get('margin_pct', 0) or 0):.1f}% margin)"
)
snap_data = [
["Revenue", f"K{float(snap.get('revenue', 0) or 0):,.0f}"],
[
"Expenses",
f"K{float(snap.get('expenses', 0) or 0):,.0f} "
f"({float(snap.get('expense_pct', 0) or 0):.0f}% of revenue)",
],
[_snap_p_label, _snap_p_value],
[
"Health Score",
f"{h_score}/100 - {_pdf_safe_text(snap.get('health_label', 'N/A'))}",
],
["Cash Flow", _pdf_safe_text(snap.get("cash_flow_status", "N/A"))],
["Data Source", _pdf_safe_text(snap.get("source", "-"))],
]
story.append(
KeepTogether(
[
_tbl(
snap_data,
[5.2 * cm, 11.8 * cm],
[
("FONTNAME", (0, 0), (0, -1), "Helvetica-Bold"),
("TEXTCOLOR", (1, 2), (1, 2), _pr_col),
("TEXTCOLOR", (1, 3), (1, 3), h_color),
],
)
]
)
)
story.append(Spacer(1, 0.3 * cm))
# โโ Expense Breakdown โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ
cats = full.get("expense_categories", [])
if cats:
story.append(_section_bar("Expense breakdown"))
story.append(Spacer(1, 0.15 * cm))
exp_data = [["Category", "Amount", "% of Expenses"]] + [
[
_pdf_safe_text(c.get("name", "")),
f"K{float(c.get('amount', 0) or 0):,.0f}",
f"{float(c.get('pct_of_expenses', 0) or 0):.1f}%",
]
for c in cats[:6]
]
story.append(
_tbl(
exp_data,
[7 * cm, 4 * cm, 4 * cm],
[
("FONTNAME", (0, 0), (-1, 0), "Helvetica-Bold"),
("BACKGROUND", (0, 0), (-1, 0), C_SLATE),
("TEXTCOLOR", (0, 0), (-1, 0), colors.white),
(
"ROWBACKGROUNDS",
(0, 1),
(-1, -1),
[colors.white, C_CARD],
),
],
)
)
story.append(Spacer(1, 0.3 * cm))
# โโ Cash Flow Analysis โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ
cf_sect = full.get("cash_flow", {})
if cf_sect.get("summary"):
story.append(_section_bar("Cash flow analysis"))
story.append(Spacer(1, 0.15 * cm))
story.append(_pdf_para(cf_sect["summary"], NM))
_ins = cf_sect.get("insight") or cf_sect.get("reasoning")
if _ins:
story.append(Spacer(1, 0.12 * cm))
story.append(
Paragraph(f"{_pdf_escape(_ins)}", SM)
)
story.append(Spacer(1, 0.3 * cm))
# โโ Loan Readiness โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ
story.append(_section_bar("Loan readiness"))
story.append(Spacer(1, 0.15 * cm))
loan_data = [
["Score", f"{l_score}/100 - {_pdf_safe_text(loan.get('status', ''))}"],
["Risk Level", _pdf_safe_text(loan.get("risk_level", "-"))],
["Safe Borrowing", f"K{float(loan.get('safe_amount', 0) or 0):,.0f}"],
["Formula", _pdf_safe_text(loan.get("formula", "-"))],
]
story.append(
_tbl(
loan_data,
[5.2 * cm, 11.8 * cm],
[
("FONTNAME", (0, 0), (0, -1), "Helvetica-Bold"),
("TEXTCOLOR", (1, 0), (1, 0), l_color),
],
)
)
improve = loan.get("improve", [])
if improve:
story.append(Spacer(1, 0.18 * cm))
story.append(Paragraph("How to improve", H3))
for item in improve:
story.append(
Paragraph(f"- {_pdf_escape(item)}", NM)
)
story.append(Spacer(1, 0.3 * cm))
# โโ Market Conditions โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ
mkt = full.get("market", {})
if mkt.get("summary"):
story.append(_section_bar("Market conditions"))
story.append(Spacer(1, 0.15 * cm))
story.append(_pdf_para(mkt["summary"], NM))
if mkt.get("opportunity"):
story.append(Spacer(1, 0.12 * cm))
story.append(
Paragraph(
f"Opportunity: {_pdf_escape(mkt['opportunity'])}",
NM,
)
)
story.append(Spacer(1, 0.3 * cm))
# โโ 6-Month Forecast โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ
fc6 = full.get("forecast_6m", {})
story.append(_section_bar("6-month forecast (base case)"))
story.append(Spacer(1, 0.15 * cm))
fc_status = (
"All 6 months profitable"
if fc6.get("all_profitable")
else f"Risk from {_pdf_safe_text(fc6.get('first_loss_month', 'month unknown'))}"
)
fc_data = [
["Month 6 profit", f"K{float(fc6.get('base_profit_m6', 0) or 0):,.0f}"],
["Total 6-mo profit", f"K{float(fc6.get('total_profit', 0) or 0):,.0f}"],
["Profitable months", f"{fc6.get('profitable_months', 0)}/6"],
["Outlook", fc_status],
]
story.append(
_tbl(
fc_data,
[6 * cm, 11 * cm],
[
("FONTNAME", (0, 0), (0, -1), "Helvetica-Bold"),
],
)
)
story.append(Spacer(1, 0.3 * cm))
# โโ Recommendations โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ
recs = full.get("recommendations", [])
if recs:
story.append(_section_bar("Top recommendations"))
story.append(Spacer(1, 0.15 * cm))
hdr = _style(
"RH",
fontSize=9,
textColor=colors.white,
fontName="Helvetica-Bold",
leading=11,
)
rec_rows = [
[
Paragraph(_pdf_escape("#"), hdr),
Paragraph(_pdf_escape("Action"), hdr),
Paragraph(_pdf_escape("Impact"), hdr),
Paragraph(_pdf_escape("Timeline"), hdr),
]
]
for i, r in enumerate(recs):
rec_rows.append(
[
Paragraph(_pdf_escape(str(i + 1)), NM_SM),
Paragraph(_pdf_escape(_pdf_safe_text(r.get("action", ""))), NM_SM),
Paragraph(_pdf_escape(_pdf_safe_text(r.get("impact", ""))), NM_SM),
Paragraph(_pdf_escape(_pdf_safe_text(r.get("timeline", ""))), NM_SM),
]
)
rt = Table(rec_rows, colWidths=[0.9 * cm, 7.3 * cm, 4.6 * cm, 4.6 * cm])
rt.setStyle(
TableStyle(
[
("BACKGROUND", (0, 0), (-1, 0), C_DARK),
("TEXTCOLOR", (0, 0), (-1, 0), colors.white),
("FONTNAME", (0, 1), (-1, -1), "Helvetica"),
("FONTSIZE", (0, 1), (-1, -1), 9),
("ROWBACKGROUNDS", (0, 1), (-1, -1), [colors.white, C_CARD]),
("GRID", (0, 0), (-1, -1), 0.35, C_GRID),
("VALIGN", (0, 0), (-1, -1), "TOP"),
("TOPPADDING", (0, 0), (-1, -1), 5),
("BOTTOMPADDING", (0, 0), (-1, -1), 5),
("LEFTPADDING", (0, 0), (-1, -1), 6),
]
)
)
story.append(rt)
story.append(Spacer(1, 0.3 * cm))
# โโ Risks โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ
risks = full.get("risks", [])
if risks:
story.append(_section_bar("Risks to watch"))
story.append(Spacer(1, 0.15 * cm))
NM_RISK = _style("RP_RISK", fontSize=10, textColor=C_AMB, leading=15)
risk_rows = [
[Paragraph(f"! {_pdf_escape(_pdf_safe_text(risk))}", NM_RISK)]
for risk in risks
]
rt2 = Table(risk_rows, colWidths=[17.4 * cm])
rt2.setStyle(
TableStyle(
[
("BACKGROUND", (0, 0), (-1, -1), colors.HexColor("#fffbeb")),
("BOX", (0, 0), (-1, -1), 0.6, colors.HexColor("#fbbf24")),
("LEFTPADDING", (0, 0), (-1, -1), 10),
("RIGHTPADDING", (0, 0), (-1, -1), 10),
("TOPPADDING", (0, 0), (-1, -1), 8),
("BOTTOMPADDING", (0, 0), (-1, -1), 8),
]
)
)
story.append(rt2)
story.append(Spacer(1, 0.3 * cm))
# โโ Footer โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ
story.append(_hr(1, C_GRID))
story.append(
Paragraph(
_pdf_escape(
"Generated by Duka AI | Powered by AMD MI300X | Qwen Model"
),
FT,
)
)
doc.build(story)
return buf.getvalue()
def build_excel_report_bytes(full: dict) -> tuple[bytes, str]:
"""Return (bytes, mime_type) for an Excel or CSV of the report data."""
from io import BytesIO, StringIO
snap = full.get("snapshot", {})
loan = full.get("loan", {})
fc6 = full.get("forecast_6m", {})
try:
import openpyxl
from openpyxl.styles import Font, PatternFill, Alignment, Border, Side
wb = openpyxl.Workbook()
# โโ Sheet 1: Overview โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ
ws = wb.active
ws.title = "Overview"
hdr_fill = PatternFill("solid", fgColor="0F172A")
hdr_font = Font(bold=True, color="FFFFFF", size=11)
sub_fill = PatternFill("solid", fgColor="1D4ED8")
sub_font = Font(bold=True, color="FFFFFF", size=10)
bold = Font(bold=True)
thin = Border(
left=Side(style="thin", color="CBD5E1"),
right=Side(style="thin", color="CBD5E1"),
top=Side(style="thin", color="CBD5E1"),
bottom=Side(style="thin", color="CBD5E1"),
)
def _hdr(ws, row, col, text, span=1):
ws.cell(row, col, text).font = hdr_font
ws.cell(row, col).fill = hdr_fill
ws.cell(row, col).alignment = Alignment(horizontal="left", vertical="center")
if span > 1:
ws.merge_cells(start_row=row, start_column=col, end_row=row, end_column=col+span-1)
def _sub(ws, row, col, text):
ws.cell(row, col, text).font = sub_font
ws.cell(row, col).fill = sub_fill
def _row(ws, row, label, value):
ws.cell(row, 1, label).font = bold
ws.cell(row, 2, value)
for c in (1, 2):
ws.cell(row, c).border = thin
r = 1
_hdr(ws, r, 1, "Duka AI โ Financial Health Report", 2); r += 1
ws.cell(r, 1, f"{full.get('business_name','')} | {full.get('location','')} | {full.get('generated_at','')}"); r += 2
_sub(ws, r, 1, "EXECUTIVE SUMMARY"); ws.cell(r, 2).fill = sub_fill; r += 1
ws.cell(r, 1, full.get("executive_summary","")).alignment = Alignment(wrap_text=True)
ws.merge_cells(start_row=r, start_column=1, end_row=r, end_column=2); r += 2
_sub(ws, r, 1, "FINANCIAL SNAPSHOT"); ws.cell(r, 2).fill = sub_fill; r += 1
_xls_profit = snap.get("profit", 0)
_xls_p_label = "Net Loss" if _xls_profit < 0 else "Profit"
_row(ws, r, "Revenue", f"K{snap.get('revenue',0):,.0f}"); r += 1
_row(ws, r, "Expenses", f"K{snap.get('expenses',0):,.0f} ({snap.get('expense_pct',0):.0f}% of revenue)"); r += 1
_row(ws, r, _xls_p_label, f"K{abs(_xls_profit):,.0f} ({snap.get('margin_pct',0):.1f}% margin)"); r += 1
_row(ws, r, "Health Score",f"{snap.get('health_score',0)}/100 โ {snap.get('health_label','')}"); r += 1
_row(ws, r, "Cash Flow", snap.get("cash_flow_status","โ")); r += 2
_sub(ws, r, 1, "LOAN READINESS"); ws.cell(r, 2).fill = sub_fill; r += 1
_row(ws, r, "Score", f"{loan.get('score',0)}/100 โ {loan.get('status','')}"); r += 1
_row(ws, r, "Risk Level", loan.get("risk_level","โ")); r += 1
_row(ws, r, "Safe Borrowing", f"K{loan.get('safe_amount',0):,.0f}"); r += 1
_row(ws, r, "Formula", loan.get("formula","โ")); r += 2
_sub(ws, r, 1, "6-MONTH FORECAST (BASE CASE)"); ws.cell(r, 2).fill = sub_fill; r += 1
outlook = "All months profitable" if fc6.get("all_profitable") else f"Risk from {fc6.get('first_loss_month','')}"
_row(ws, r, "Month 6 Profit", f"K{fc6.get('base_profit_m6',0):,.0f}"); r += 1
_row(ws, r, "Total 6-Mo Profit", f"K{fc6.get('total_profit',0):,.0f}"); r += 1
_row(ws, r, "Outlook", outlook); r += 2
# Recommendations
recs = full.get("recommendations", [])
if recs:
_sub(ws, r, 1, "TOP RECOMMENDATIONS"); ws.cell(r, 2).fill = sub_fill; ws.cell(r, 3).fill = sub_fill; r += 1
ws.cell(r, 1, "Action").font = bold; ws.cell(r, 2, "Impact").font = bold; ws.cell(r, 3, "Timeline").font = bold; r += 1
for rec in recs:
ws.cell(r, 1, rec.get("action",""))
ws.cell(r, 2, rec.get("impact",""))
ws.cell(r, 3, rec.get("timeline",""))
r += 1
r += 1
# Risks
risks = full.get("risks", [])
if risks:
_sub(ws, r, 1, "RISKS TO WATCH"); ws.cell(r, 2).fill = sub_fill; r += 1
for risk in risks:
ws.cell(r, 1, f"โ {risk}"); r += 1
r += 1
ws.column_dimensions["A"].width = 28
ws.column_dimensions["B"].width = 40
ws.column_dimensions["C"].width = 18
# โโ Sheet 2: Expense Breakdown โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ
cats = full.get("expense_categories", [])
if cats:
ws2 = wb.create_sheet("Expense Breakdown")
_hdr(ws2, 1, 1, "Category")
_hdr(ws2, 1, 2, "Amount (K)")
_hdr(ws2, 1, 3, "% of Expenses")
for i, c in enumerate(cats, start=2):
ws2.cell(i, 1, c.get("name",""))
ws2.cell(i, 2, round(c.get("amount",0), 2))
ws2.cell(i, 3, f"{c.get('pct_of_expenses',0):.1f}%")
ws2.column_dimensions["A"].width = 28
ws2.column_dimensions["B"].width = 18
ws2.column_dimensions["C"].width = 18
out = BytesIO()
wb.save(out)
return out.getvalue(), "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet"
except ImportError:
# openpyxl not installed โ produce a UTF-8 CSV instead
import csv
sio = StringIO()
w = csv.writer(sio)
w.writerow(["Duka AI โ Financial Health Report"])
w.writerow([full.get("business_name",""), full.get("location",""), full.get("generated_at","")])
w.writerow([])
w.writerow(["EXECUTIVE SUMMARY"])
w.writerow([full.get("executive_summary","")])
w.writerow([])
w.writerow(["FINANCIAL SNAPSHOT", ""])
_csv_profit = snap.get("profit", 0)
_csv_p_label = "Net Loss" if _csv_profit < 0 else "Profit"
w.writerow(["Revenue", f"K{snap.get('revenue',0):,.0f}"])
w.writerow(["Expenses", f"K{snap.get('expenses',0):,.0f}"])
w.writerow([_csv_p_label, f"K{abs(_csv_profit):,.0f}"])
w.writerow(["Health Score",f"{snap.get('health_score',0)}/100"])
w.writerow([])
w.writerow(["LOAN READINESS", ""])
w.writerow(["Score", f"{loan.get('score',0)}/100"])
w.writerow(["Safe Borrowing",f"K{loan.get('safe_amount',0):,.0f}"])
w.writerow([])
w.writerow(["RECOMMENDATIONS", "Impact", "Timeline"])
for r in full.get("recommendations",[]):
w.writerow([r.get("action",""), r.get("impact",""), r.get("timeline","")])
w.writerow([])
w.writerow(["RISKS TO WATCH"])
for risk in full.get("risks",[]):
w.writerow([risk])
return sio.getvalue().encode("utf-8"), "text/csv"
def render_generate_report_page() -> None:
render_page_header(
"Generate Report",
"AI-compiled financial health report โ ready to share with your bank or accountant.",
"Report",
)
report = require_report()
if not report:
return
baseline = st.session_state.get("baseline") or get_baseline(st.session_state)
if not baseline:
st.warning("No financial baseline found. Run an analysis in Chat Advisor first.")
return
metrics = st.session_state.get("metrics") or compute_metrics(baseline)
bp = {
"business_type": st.session_state.get("business_type", ""),
"location": st.session_state.get("location", ""),
"products_services": st.session_state.get("products_services", ""),
}
# โโ Trigger / cache logic โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ
full: dict | None = st.session_state.get("generated_report")
if full is None:
if not st.session_state.get("_report_generating"):
st.info(
"Click below to compile a full financial health report from your analysis data. "
"The AI will write the narrative; all numbers come from verified calculations."
)
if st.button("โก Generate Full Report", type="primary", use_container_width=True):
st.session_state["_report_generating"] = True
st.rerun()
return
# Generating โ phased progress (matches agents/report_agent.generate_full_report)
from agents.report_agent import generate_full_report
with st.status("๐ Starting financial report pipelineโฆ", expanded=True) as status:
st.markdown(
"**Typical duration:** about **15โ60 seconds**, depending on API latency. "
"Every Kwacha amount is taken from Python-verified calculations first; "
"the AI only writes narrative text grounded in those numbers."
)
st.caption(
"Pipeline: verified snapshot โ expense categories โ 6-month base-case forecast "
"โ AI executive summary โ AI recommendations & risks โ merge for PDF/Excel export."
)
prog = st.progress(0)
def _on_report_progress(step: int, total: int, detail: str) -> None:
pct = min(step / max(total, 1), 1.0)
try:
prog.progress(pct, text=f"Step {step}/{total}")
except TypeError:
prog.progress(pct)
short = detail if len(detail) <= 80 else detail[:77] + "โฆ"
status.update(label=f"๐ {short}", state="running")
st.markdown(f"##### Step {step} of {total}")
st.markdown(detail)
st.divider()
try:
full = generate_full_report(
baseline, metrics, report, bp, on_progress=_on_report_progress
)
except Exception as exc:
st.session_state["_report_generating"] = False
try:
prog.progress(0, text="Stopped")
except TypeError:
prog.progress(0)
status.update(
label="โ Report generation failed โ see details below",
state="error",
expanded=True,
)
st.exception(exc)
return
try:
prog.progress(1.0, text="Done")
except TypeError:
prog.progress(1.0)
status.update(
label="โ
Report compiled โ scroll down for preview and downloads.",
state="complete",
expanded=False,
)
st.session_state["generated_report"] = full
st.session_state["_report_generating"] = False
st.rerun()
# โโ Action bar (top) โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ
snap = full.get("snapshot", {})
loan_d = full.get("loan", {})
fc6 = full.get("forecast_6m", {})
pdf_bytes = build_full_report_pdf(full)
excel_bytes, xls_mime = build_excel_report_bytes(full)
xls_ext = "xlsx" if "spreadsheet" in xls_mime else "csv"
xls_label = "๐ Download Excel" if xls_ext == "xlsx" else "๐ Download CSV"
col_pdf, col_xls, col_regen = st.columns([3, 3, 2], gap="small")
col_pdf.download_button(
"๐ Download PDF", pdf_bytes, "duka-ai-report.pdf", "application/pdf",
use_container_width=True, type="primary", key="rpt_dl_pdf_top",
)
col_xls.download_button(
xls_label, excel_bytes, f"duka-ai-report.{xls_ext}", xls_mime,
use_container_width=True, key="rpt_dl_xls_top",
)
if col_regen.button("๐ Regenerate", use_container_width=True):
st.session_state.pop("generated_report", None)
st.rerun()
st.markdown("", unsafe_allow_html=True)
_rn = _report_text_escape(full.get("business_name", "Your Business"))
_rloc = _report_text_escape(full.get("location", "Zambia"))
_rat = _report_text_escape(full.get("generated_at", ""))
_snap_pr = float(snap.get("profit", 0) or 0)
_kpi_pc = "loss" if _snap_pr < 0 else "gain"
st.markdown(
''
'',
unsafe_allow_html=True,
)
# โโ Hero + KPI ribbon โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ
st.markdown(
f"""
Financial Health Report
{_rn}
{_rloc} ยท Generated {_rat}
Verified numbers ยท AI narrative
Monthly revenue
K{snap.get("revenue", 0):,.0f}
Monthly expenses
K{snap.get("expenses", 0):,.0f}
Net profit
K{abs(_snap_pr):,.0f}
""",
unsafe_allow_html=True,
)
st.markdown(
f"""
Executive summary
{_report_text_escape(full.get("executive_summary", "โ"))}
""",
unsafe_allow_html=True,
)
# โโ Financial Snapshot + Expense Breakdown โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ
def _score_color(score: int) -> str:
return "#10B981" if score >= 70 else ("#F59E0B" if score >= 40 else "#EF4444")
h_color = _score_color(snap.get("health_score", 0))
l_color = _score_color(loan_d.get("score", 0))
col_snap, col_exp = st.columns([1, 1], gap="medium")
with col_snap:
_cfs = _report_text_escape(snap.get("cash_flow_status", "โ"))
_hl = _report_text_escape(snap.get("health_label", "N/A"))
_src = _report_text_escape(snap.get("source", "โ"))
_pr_disp = float(snap.get("profit", 0) or 0)
_pr_col = "#FCA5A5" if _pr_disp < 0 else "#6EE7B7"
st.markdown(
f"""
Financial snapshot
| Revenue |
K{snap.get("revenue",0):,.0f} |
| Expenses |
K{snap.get("expenses",0):,.0f}
({snap.get("expense_pct",0):.0f}% of rev.)
|
| Profit |
K{_pr_disp:,.0f}
({snap.get("margin_pct",0):.1f}% margin)
|
|
| Health score |
{snap.get("health_score",0)}/100 โ {_hl}
|
| Cash flow |
{_cfs} |
| Data source |
{_src} |
""",
unsafe_allow_html=True,
)
with col_exp:
cats = full.get("expense_categories", [])
if cats:
rows_html = "".join(
f"""
| {i + 1}. {_report_text_escape(c.get("name", ""))} |
K{c.get("amount", 0):,.0f} |
{c.get("pct_of_expenses", 0):.1f}% |
"""
for i, c in enumerate(cats[:6])
)
st.markdown(
f"""
""",
unsafe_allow_html=True,
)
else:
st.markdown(
'
'
"No expense category breakdown available for this analysis."
"
",
unsafe_allow_html=True,
)
st.markdown("
", unsafe_allow_html=True)
# โโ Cash Flow + Loan side by side โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ
col_cf, col_loan = st.columns([1, 1], gap="medium")
with col_cf:
cf_sect = full.get("cash_flow", {})
_cf_sum = _report_text_escape(cf_sect.get("summary", "โ"))
_cf_ins = cf_sect.get("insight") or cf_sect.get("reasoning") or ""
_cf_ins_html = (
f"
{_report_text_escape(_cf_ins)}
"
if _cf_ins
else ""
)
st.markdown(
f"""
Cash flow analysis
{_cf_sum}
{_cf_ins_html}
""",
unsafe_allow_html=True,
)
with col_loan:
improve_html = "".join(
f'
{_report_text_escape(item)}'
for item in loan_d.get("improve", [])
)
_lst = _report_text_escape(loan_d.get("status", "โ"))
_lfr = _report_text_escape(loan_d.get("formula", ""))
st.markdown(
f"""
Loan readiness
{loan_d.get("score", 0)}/100 โ {_lst}
Safe borrowing: K{loan_d.get("safe_amount", 0):,.0f}
ยท {_lfr}
{"
" if improve_html else ""}
""",
unsafe_allow_html=True,
)
st.markdown("
", unsafe_allow_html=True)
# โโ 6-Month outlook โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ
_fc_out = (
"All 6 months profitable (base case)."
if fc6.get("all_profitable")
else f"Attention: losses possible from {fc6.get('first_loss_month', 'โ')} upward."
)
st.markdown(
f"""
6-month outlook (base case)
Month 6 profit
K{fc6.get("base_profit_m6", 0):,.0f}
Total 6-mo profit
K{fc6.get("total_profit", 0):,.0f}
Profitable months
{fc6.get("profitable_months", 0)}/6
Outlook
{_report_text_escape(_fc_out)}
""",
unsafe_allow_html=True,
)
st.markdown("
", unsafe_allow_html=True)
# โโ Market Conditions โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ
mkt = full.get("market", {})
if mkt.get("summary"):
monitor_html = "".join(
f'
'
f'{_report_text_escape(m)}'
for m in mkt.get("monitor", [])
)
_opp = mkt.get("opportunity", "")
_opp_blk = (
f"
Opportunity: {_report_text_escape(_opp)}
"
if _opp
else ""
)
st.markdown(
f"""
Market conditions
{_report_text_escape(mkt.get("summary", "โ"))}
{_opp_blk}
{("
" + monitor_html + "
") if monitor_html else ""}
""",
unsafe_allow_html=True,
)
# โโ Recommendations โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ
recs = full.get("recommendations", [])
if recs:
TIMELINE_COLORS = {"Now": "#10B981", "This month": "#2563EB"}
rec_cards = ""
for i, r in enumerate(recs):
t_color = TIMELINE_COLORS.get(r.get("timeline", ""), "#64748B")
_act = _report_text_escape(r.get("action", "โ"))
_imp = _report_text_escape(r.get("impact", "โ"))
_tl = _report_text_escape(r.get("timeline", "โ"))
rec_cards += f"""
"""
st.markdown(
f"""
Top recommendations
{rec_cards}
""",
unsafe_allow_html=True,
)
# โโ Risks to Watch โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ
risks = full.get("risks", [])
if risks:
risk_html = "".join(
f'
'
f'โ '
f'{_report_text_escape(r)}
'
for r in risks
)
st.markdown(
f"""
Risks to watch
{risk_html}
""",
unsafe_allow_html=True,
)
# โโ Footer / branding โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ
st.markdown(
'"
"
'
'Share this report with your bank manager or accountant via WhatsApp or email. '
'Download the PDF above and attach it to your message.'
'
',
unsafe_allow_html=True,
)
# Plain-text copy of the report
report_text = (
f"Duka AI โ Financial Health Report\n"
f"{full.get('business_name','')} | {full.get('location','')} | {full.get('generated_at','')}\n"
f"{'='*60}\n\n"
f"EXECUTIVE SUMMARY\n{full.get('executive_summary','')}\n\n"
f"FINANCIAL SNAPSHOT\n"
f"Revenue: K{snap.get('revenue',0):,.0f}\n"
f"Expenses: K{snap.get('expenses',0):,.0f} ({snap.get('expense_pct',0):.0f}%)\n"
f"{'Net Loss' if snap.get('profit',0) < 0 else 'Profit'}: K{abs(snap.get('profit',0)):,.0f} ({snap.get('margin_pct',0):.1f}% margin)\n"
f"Health Score: {snap.get('health_score',0)}/100 โ {snap.get('health_label','')}\n\n"
f"LOAN READINESS\n"
f"Score: {loan_d.get('score',0)}/100 โ {loan_d.get('status','')}\n"
f"Safe borrowing: K{loan_d.get('safe_amount',0):,.0f}\n\n"
+ "TOP RECOMMENDATIONS\n"
+ "\n".join(
f"{i+1}. {r.get('action','')} โ {r.get('impact','')} ({r.get('timeline','')})"
for i, r in enumerate(recs)
) + "\n\n"
+ ("RISKS TO WATCH\n" + "\n".join(f"โ {r}" for r in risks) + "\n\n" if risks else "")
+ f"{'='*60}\nGenerated by Duka AI | Powered by AMD MI300X | Qwen Model"
)
st.text_area("Copy report text", report_text, height=200, key="report_share_text")
sh_pdf, sh_xls = st.columns(2)
sh_pdf.download_button(
"๐ Download PDF",
pdf_bytes,
"duka-ai-report.pdf",
"application/pdf",
use_container_width=True,
key="rpt_dl_pdf_share",
)
sh_xls.download_button(
xls_label,
excel_bytes,
f"duka-ai-report.{xls_ext}",
xls_mime,
use_container_width=True,
key="rpt_dl_xls_share",
)
# โโ Schedule automatic reports โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ
st.markdown("", unsafe_allow_html=True)
st.markdown(
"""
📬 Want this report automatically?
I can analyze your business and email you a complete financial report on a
schedule you choose — no action needed from you.
""",
unsafe_allow_html=True,
)
st.markdown(
""
"Set up automatic reports:
",
unsafe_allow_html=True,
)
sched_freq = st.session_state.get("schedule_frequency")
btn_w, btn_m, btn_c = st.columns(3, gap="small")
if btn_w.button(
"๐
Weekly โ every Monday",
use_container_width=True,
type="primary" if sched_freq == "weekly" else "secondary",
key="sched_weekly",
):
st.session_state["schedule_frequency"] = "weekly"
st.session_state.pop("schedule_confirmed", None)
st.rerun()
if btn_m.button(
"๐
Monthly โ 1st of month",
use_container_width=True,
type="primary" if sched_freq == "monthly" else "secondary",
key="sched_monthly",
):
st.session_state["schedule_frequency"] = "monthly"
st.session_state.pop("schedule_confirmed", None)
st.rerun()
if btn_c.button(
"โ๏ธ Custom schedule",
use_container_width=True,
type="primary" if sched_freq == "custom" else "secondary",
key="sched_custom",
):
st.session_state["schedule_frequency"] = "custom"
st.session_state.pop("schedule_confirmed", None)
st.rerun()
if sched_freq and not st.session_state.get("schedule_confirmed"):
st.markdown("", unsafe_allow_html=True)
if sched_freq == "custom":
_custom_map = {
"Every 2 weeks (bi-weekly)": "biweekly",
"Quarterly (every 3 months)": "quarterly",
}
_chosen = st.selectbox(
"Choose your schedule:",
list(_custom_map.keys()),
key="sched_custom_choice",
)
effective_freq = _custom_map[_chosen]
else:
effective_freq = sched_freq
sched_email = st.text_input(
"Your email address",
placeholder="you@example.com",
key="sched_email_input",
)
if st.button(
"Set up automatic reports โ
",
type="primary",
use_container_width=True,
key="sched_confirm",
):
import re as _re
if not sched_email or not _re.match(r"[^@]+@[^@]+\.[^@]+", sched_email.strip()):
st.error("Please enter a valid email address.")
else:
_biz_ctx = {
"business_type": st.session_state.get("business_type", ""),
"location": st.session_state.get("location", "Zambia"),
"products_services": st.session_state.get("products_services", ""),
"manual_notes": st.session_state.get("manual_notes", ""),
"manual_revenue": st.session_state.get("manual_revenue", 0.0),
"manual_expenses": st.session_state.get("manual_expenses", 0.0),
"manual_debt": st.session_state.get("manual_debt", 0.0),
"manual_staff": st.session_state.get("manual_staff", 0),
}
try:
from tools.scheduler import save_schedule
_sid = save_schedule(
email=sched_email.strip(),
frequency=effective_freq,
business_context=_biz_ctx,
)
st.session_state["schedule_confirmed"] = True
st.session_state["schedule_email"] = sched_email.strip()
st.session_state["schedule_id"] = _sid
st.rerun()
except Exception as _exc:
st.error(f"Could not save schedule: {_exc}")
if st.session_state.get("schedule_confirmed"):
_conf_email = st.session_state.get("schedule_email", "")
_conf_freq = st.session_state.get("schedule_frequency", "weekly")
_freq_display = {
"weekly": "every Monday at 8 AM",
"monthly": "on the 1st of each month at 8 AM",
"biweekly": "every 2 weeks at 8 AM",
"quarterly": "every quarter at 8 AM",
"custom": "on your chosen schedule",
}.get(_conf_freq, "on your chosen schedule")
st.success(
f"Scheduled! Your financial report will be emailed to **{_conf_email}** {_freq_display}."
)
if st.button("Cancel / change schedule", key="sched_cancel"):
for _k in ["schedule_confirmed", "schedule_frequency", "schedule_email", "schedule_id"]:
st.session_state.pop(_k, None)
st.rerun()
def _market_intel_identity_reply() -> str:
return (
"I'm **Duka AI's Market Intelligence Agent** โ part of **Duka AI**, built for **Zambian SMEs**. "
"I connect **your business profile and numbers** from your last analysis to **market conditions** "
"(this pageโs snapshot and, when enabled, web sources from when your report was built). "
"I'm not a general chatbot for every topic โ I stay focused on **your market and your financial picture**.\n\n"
"Try: *What's the biggest risk for my type of business right now?* or "
"*How does my revenue pattern line up with what the market intel says?*"
)
def _build_market_chat_business_context(report: dict[str, Any]) -> str:
"""Ground Market Intel chat in the same numbers as the rest of the report."""
lines: list[str] = []
cf = report.get("cashflow") or {}
lines.append("VERIFIED BUSINESS NUMBERS (from last analysis โ cite these for sales/profit/trend questions):\n")
lines.append(
f"- Revenue K{float(cf.get('revenue', 0)):,.0f} ยท Expenses K{float(cf.get('expenses', 0)):,.0f} ยท "
f"Profit K{float(cf.get('profit', 0)):,.0f} ยท Margin {float(cf.get('profit_margin', 0)):.1f}% ยท "
f"Status: {cf.get('cash_flow_status', 'n/a')}\n"
)
if (cf.get("summary") or "").strip():
lines.append(f"- Summary: {(cf.get('summary') or '')[:500]}\n")
parsed = report.get("parsed_data") or {}
ac = parsed.get("_analysis_context") or {}
doc = report.get("document_analysis") or ac.get("document_analysis") or {}
bp = report.get("business_profile") or ac.get("business_profile") or {}
bt = (parsed.get("business_type") or bp.get("business_type") or "").strip()
loc = (parsed.get("location") or bp.get("location") or "").strip()
products = (parsed.get("products_services") or bp.get("products_services") or "").strip()
if bt or loc:
lines.append(f"- Business: {bt or 'SME'} ยท Location: {loc or 'Zambia'}\n")
if products:
lines.append(f"- Products/services listed by owner: {products[:600]}\n")
mb = doc.get("monthly_breakdown") if isinstance(doc.get("monthly_breakdown"), list) else []
if len(mb) > 1:
lines.append(
"MONTHLY REVENUE PATTERN (use this for โtrendingโ, seasonality, up/down โ not individual SKUs):\n"
)
for entry in mb[:14]:
if not isinstance(entry, dict):
continue
mn = entry.get("month") or entry.get("Month") or ""
rev = entry.get("revenue", entry.get("Revenue"))
if rev is not None:
try:
lines.append(f" ยท {mn}: revenue K{float(rev):,.0f}\n")
except (TypeError, ValueError):
pass
eb = parsed.get("expenses_breakdown") or {}
if isinstance(eb, dict) and eb:
top = sorted(
((k, float(v or 0)) for k, v in eb.items() if float(v or 0) > 0),
key=lambda x: x[1],
reverse=True,
)[:5]
if top:
lines.append(
"TOP EXPENSE CATEGORIES (if user asks where money goes):\n"
+ "".join(f" ยท {k}: K{v:,.0f}\n" for k, v in top)
)
ts = report.get("transaction_summary") or {}
if ts.get("average_daily_sales"):
lines.append(
f"- Transactions (if connected): avg daily sales ~{format_currency(ts['average_daily_sales'])}, "
f"risk {ts.get('cash_flow_risk', 'n/a')}.\n"
)
lines.append(
"\nSCOPE NOTE: There is usually **no per-product SKU sales feed**. "
"For โare my products trendingโ, use **monthly revenue trend** above + ownerโs product list + market intel. "
"If monthly data is missing, say revenue is only known as totals above.\n"
)
return "".join(lines)[:7500]
def _classify_market_chat_input(question: str) -> str | None:
"""Return a canned reply key, or None to use the LLM."""
low = (question or "").lower().strip()
if not low:
return "empty"
if any(
p in low
for p in (
"who are you",
"what are you",
"what do you do",
"introduce yourself",
)
):
return "identity"
if low in ("thanks", "thank you", "ty", "cheers", "ta"):
return "thanks"
return None
def _handle_market_question(
question: str,
market: dict,
chat_history: list[dict],
report: dict[str, Any] | None = None,
) -> str:
"""Answer using the market block + verified business numbers from the same report."""
from agents import request_llm_chat
canned = _classify_market_chat_input(question)
if canned == "identity":
return _market_intel_identity_reply()
if canned == "thanks":
return "Glad it helped โ ask anything else about your market snapshot or your numbers."
if canned == "empty":
return "Type a question about your market or how it relates to your business."
sources = market.get("sources", [])
queries = market.get("queries_used", [])
web_used = market.get("web_search_used", False)
# Build context from real search data
source_context = ""
if web_used and sources:
source_context = "\n\nWEB SEARCH SOURCES (real, retrieved data โ use these):\n"
seen: set[str] = set()
for s in sources:
url = s.get("url", "")
if url in seen:
continue
seen.add(url)
source_context += f"- {s.get('title', '')} ({url})\n"
if s.get("content"):
source_context += f" {s['content'][:300]}\n"
market_context = (
f"MARKET INTELLIGENCE RESULTS:\n"
f"Business type: {market.get('business_type', 'SME')}\n"
f"Location: {market.get('location', 'Zambia')}\n"
f"Market summary: {market.get('market_summary', '')}\n"
f"Key risks: {'; '.join(market.get('risk_factors', []))}\n"
f"Opportunity: {market.get('opportunity', '')}\n"
f"Recommendation: {market.get('recommendation', '')}\n"
f"What to monitor: {'; '.join(market.get('monitor_next', []))}\n"
f"Search queries used: {'; '.join(queries)}\n"
f"{source_context}"
)
biz_block = _build_market_chat_business_context(report) if report else ""
if web_used:
grounding = (
"Live Tavily web search was used when this report was built โ snippets and titles below are real. "
"Answer using the MARKET INTELLIGENCE block + VERIFIED BUSINESS NUMBERS + web excerpts. "
"Do NOT invent statistics. Cite or paraphrase sources when you mention external facts."
)
else:
grounding = (
"No live web search ran for this report (missing TAVILY_API_KEY or search failed). "
"The MARKET INTELLIGENCE RESULTS below are best-effort synthesis from business context โ "
"do NOT claim you ran a fresh web search. "
"Still use VERIFIED BUSINESS NUMBERS for anything about **their** sales, profit, or month-to-month trend.\n"
"If something isn't in the data, say exactly what's missing (e.g. no SKU-level sales)."
)
system = (
"You are **Duka AI's Market Intelligence Agent** inside **Duka AI** for **Zambian SMEs**. "
"You are NOT a generic assistant for โthe economyโ or random topics โ you tie **market conditions** to "
"**this owner's** business type, location, and **verified numbers**.\n\n"
"WHO YOU ARE (if asked): You help this user interpret **their** market snapshot (this page) and how it "
"relates to **their revenue/cash flow** โ not broad unrelated advice.\n\n"
"PRODUCT / โTRENDINGโ QUESTIONS: You normally do **not** have per-SKU sales. "
"Answer using **monthly revenue** (if provided), overall totals, their stated products/services, "
"and the market intel risks/opportunities. Say clearly when you're inferring vs when data proves it.\n\n"
f"{grounding}\n"
"Be concise (2โ6 sentences unless they ask for detail). "
"Never refuse to discuss trends if monthly revenue or totals exist โ use them.\n\n"
f"{biz_block}\n"
f"{market_context}"
)
messages = [
{"role": "system", "content": system},
*[{"role": m["role"], "content": m["content"]} for m in chat_history[-8:]],
{"role": "user", "content": question},
]
result = request_llm_chat(messages, temperature=0.2, max_tokens=500)
return result or "I couldn't generate a response right now. Please try again."
def render_market_intel_page() -> None:
render_page_header(
"Market Intel",
"Live web search for Zambian market conditions, risks, and opportunities.",
"Live",
)
report = require_report()
if not report:
return
market = report.get("market", {})
st.markdown(
'',
unsafe_allow_html=True,
)
# Two-column layout: intel card left, chat panel right
col_intel, col_chat = st.columns([3, 2], gap="large")
with col_intel:
# Market data was computed during Chat Advisor analysis โ we don't re-search here.
render_market_intelligence_section(report)
with col_chat:
web_used = market.get("web_search_used", False)
sources = market.get("sources", [])
if web_used:
sub_line = (
f"Grounded in {len(sources)} live web sources ยท "
"Ask anything about your market"
)
else:
sub_line = "Ask anything about your business's market conditions in Zambia"
st.markdown(
'',
unsafe_allow_html=True,
)
st.markdown(
''
'
'
'๐ Ask the Market Agent'
''
f"{html.escape(sub_line)}"
""
"
",
unsafe_allow_html=True,
)
# Suggestion pills
sugg_pills = [
"๐ What's the market trend?",
"โ ๏ธ What are the main risks?",
"๐ฏ What's my best opportunity?",
"๐ฐ How do prices affect me?",
"๐๏ธ What's the best season to grow?",
"๐ How do I beat competition?",
]
p1, p2 = st.columns(2)
for idx, pill in enumerate(sugg_pills):
col = p1 if idx % 2 == 0 else p2
if col.button(pill, key=f"mpill_{idx}", use_container_width=True):
st.session_state["market_question"] = pill
st.markdown("
", unsafe_allow_html=True)
# Chat history (same card shell as Cash Flow Forecast conversation)
if "market_chat" not in st.session_state:
st.session_state["market_chat"] = []
if not st.session_state["market_chat"]:
st.markdown(
'
No messages yet โ tap a suggestion '
"or type below.
",
unsafe_allow_html=True,
)
for msg in st.session_state["market_chat"]:
if msg["role"] == "user":
escaped = (
msg["content"]
.replace("&", "&")
.replace("<", "<")
.replace(">", ">")
)
st.markdown(
f'
',
unsafe_allow_html=True,
)
else:
with st.chat_message("assistant", avatar="๐"):
st.markdown(msg["content"])
st.markdown("
⚡ AI-powered · 🌎 Built for Africa
Your AI financial advisor
for African businesses
Analyze cash flow, loan readiness, and market conditions.
Talk to your finances like a conversation, not a spreadsheet.
""",
unsafe_allow_html=True,
)
if st.session_state.pop("scroll_to_demo_analyze", False):
components.html(
"""
""",
height=0,
width=0,
)
if (
st.session_state.get("pending_demo_analysis_run")
and not st.session_state.get("analysis_report")
):
with st.status("โก Demo analysis in progressโฆ", expanded=True):
st.write("๐ Sample income statement loaded ยท **Lusaka Grocery Shop** profile applied.")
st.write("๐ง Starting all agents next โ detailed steps appear in the panel below.")
st.caption("Stay on this page โ the workspace scrolls to **Run Full Business Analysis** automatically.")
# โโ Powered-by indicator โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ
_provider = os.getenv("LLM_PROVIDER", "amd").strip().upper()
_model_display = (
"AMD MI300X"
if _provider == "AMD"
else ("OpenAI-compatible" if _provider == "OPENAI" else _provider or "AMD MI300X")
)
st.markdown(
f"""
Powered by Duka AI · {_model_display}
· Talk to your finances like a conversation
""",
unsafe_allow_html=True,
)
# โโ Pre-fill banner & scroll (welcome sample uses numbers anchor + hides duplicate example pills)
_did_prefill = st.session_state.pop("form_prefilled", False)
_scroll_numbers = st.session_state.pop("scroll_to_numbers", False)
if _did_prefill:
if st.session_state.get("hide_example_prompts"):
if st.session_state.get("pending_demo_analysis_run"):
st.success(
"โ
**Lusaka Grocery Shop** demo loaded โ sample income statement + figures are applied. "
"**Running full analysis automatically** โ watch the progress panel below."
)
else:
st.success(
"โ
**Lusaka Grocery Shop** sample loaded โ your story and figures are filled in above. "
"Upload a file below (optional) or jump to **Manual numbers**, then click **Run Full Business Analysis**."
)
else:
st.success(
"โ
Sample business loaded โ review the details below and click **Run Full Business Analysis**"
)
if st.session_state.get("pending_demo_analysis_run"):
_scroll_id = "duka-ai-analyze-anchor"
else:
_scroll_id = "duka-ai-numbers-anchor" if _scroll_numbers else "duka-ai-form-anchor"
components.html(
f"""
""",
height=0,
width=0,
)
elif _scroll_numbers:
components.html(
"""
""",
height=0,
width=0,
)
has_existing_analysis = bool(st.session_state.get("analysis_report"))
if not has_existing_analysis:
# โโ Section 1: Business Context โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ
st.markdown('', unsafe_allow_html=True)
render_section_intro(
"Tell us about your business",
"A few details help Duka AI tailor its analysis to your situation.",
)
context_col_1, context_col_2, context_col_3, context_col_4 = st.columns(4, gap="medium")
with context_col_1:
st.session_state.business_type = st.text_input(
"Business type",
value=st.session_state.business_type,
placeholder="Grocery shop",
help="What kind of business do you run?",
)
with context_col_2:
st.session_state.location = st.text_input(
"Location",
value=st.session_state.location,
placeholder="Lusaka",
help="City or area where your business operates",
)
with context_col_3:
st.session_state.products_services = st.text_input(
"Main products / services",
value=st.session_state.products_services,
placeholder="Groceries, mealie meal, drinks",
help="What do you sell or offer?",
)
with context_col_4:
st.session_state.main_question = st.text_input(
"Main question or concern",
value=st.session_state.main_question,
placeholder="Should I restock or save cash?",
help="What do you want Duka AI to help with?",
)
st.session_state.manual_notes = st.text_area(
"Additional notes (optional)",
value=st.session_state.manual_notes,
placeholder="I run a small grocery shop in Lusaka. I want to know whether to restock or save cash. This week I made K4,500 in sales...",
help="Add any extra details, financial numbers, debt, or business notes.",
)
st.markdown('', unsafe_allow_html=True)
continue_to_data = st.button(
"โ Continue to data sources",
key="continue_to_data_sources",
use_container_width=True,
help="Jump to the upload/connect/manual section",
)
st.markdown(
'Want higher accuracy? Add a document or figures in the section below first.
',
unsafe_allow_html=True,
)
if continue_to_data:
components.html(
"""
""",
height=0,
width=0,
)
st.divider()
if not st.session_state.get("hide_example_prompts"):
st.markdown('Example prompts
', unsafe_allow_html=True)
render_example_prompt_pills(sample_cases)
# โโ Trust strip โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ
st.markdown("
", unsafe_allow_html=True)
_trust_items = [
("๐", "Bank-grade security", "Your data stays private"),
("๐ค", "4 AI agents", "Working together for you"),
("๐", "Built for Africa", "Zambian market context"),
("โก", "AMD MI300X", "Enterprise-grade GPU power"),
]
for _col, (_icon, _title, _sub) in zip(st.columns(4), _trust_items):
with _col:
st.markdown(
f"""
{_icon}
{_title}
{_sub}
""",
unsafe_allow_html=True,
)
st.markdown("
", unsafe_allow_html=True)
st.divider()
# โโ Section 2: Data Sources โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ
st.markdown('', unsafe_allow_html=True)
render_section_intro(
"๐ Want more precise analysis? Add your numbers",
"Choose one path below. Any extra numbers help Duka AI give tighter, more specific advice.",
)
uploaded_analysis: dict[str, Any] | None = None
upload_error: str | None = None
option_cols = st.columns(3, gap="small")
for column, (label, mode_value) in zip(option_cols, NUMBER_SOURCE_OPTIONS):
with column:
if st.button(label, key=f"numbers_mode_{mode_value}"):
st.session_state.numbers_entry_mode = mode_value
st.rerun()
selected_mode = st.session_state.numbers_entry_mode
st.markdown(
'You can switch between options at any time before you analyze.
',
unsafe_allow_html=True,
)
if selected_mode == "manual":
st.markdown('Enter only the figures you know. Even partial numbers make the analysis more precise.
', unsafe_allow_html=True)
num_col1, num_col2, num_col3, num_col4 = st.columns(4, gap="medium")
with num_col1:
st.session_state.manual_revenue = st.number_input(
"Weekly or monthly revenue (K)",
min_value=0.0, value=float(st.session_state.manual_revenue), step=100.0, format="%.0f",
help="Your typical sales or income per week or month in Kwacha",
)
with num_col2:
st.session_state.manual_expenses = st.number_input(
"Monthly expenses (K)",
min_value=0.0, value=float(st.session_state.manual_expenses), step=100.0, format="%.0f",
help="Your total monthly costs โ stock, rent, wages, etc.",
)
with num_col3:
st.session_state.manual_debt = st.number_input(
"Outstanding debt (K)",
min_value=0.0, value=float(st.session_state.manual_debt), step=100.0, format="%.0f",
help="Any loans or supplier debt you currently owe",
)
with num_col4:
st.session_state.manual_staff = st.number_input(
"Number of staff",
min_value=0, value=int(st.session_state.manual_staff), step=1,
help="How many people work in your business (including yourself)",
)
elif selected_mode == "upload":
st.markdown('Upload your business document
', unsafe_allow_html=True)
st.markdown('Select the document type, upload the file, and Duka AI will extract key financial figures.
', unsafe_allow_html=True)
st.caption("Accepted file types: CSV, Excel (.xlsx), TXT, PDF")
st.caption("PDF support is experimental. For best results, upload CSV, Excel, or text records.")
st.session_state.document_type = st.selectbox(
"Document type",
DOCUMENT_TYPES,
index=DOCUMENT_TYPES.index(st.session_state.document_type) if st.session_state.document_type in DOCUMENT_TYPES else 0,
)
uploaded_file = st.file_uploader(
"Upload business document",
type=["csv", "xlsx", "txt", "pdf"],
help="Upload an income statement, sales record, expense log, bank statement, mobile money statement, or business notes.",
)
if uploaded_file is not None:
try:
_doc_fp = f"{uploaded_file.name}:{uploaded_file.size}"
_is_new_file = st.session_state.get("_doc_fingerprint") != _doc_fp
if _is_new_file:
_ext = uploaded_file.name.rsplit(".", 1)[-1].upper() if "." in uploaded_file.name else "FILE"
with st.status(
f"๐ Reading {uploaded_file.name} โฆ",
expanded=True,
) as _upload_status:
st.write(f"๐ Detecting document structure ({_ext} format)โฆ")
st.write("๐ Extracting financial figuresโฆ")
uploaded_analysis = parse_uploaded_file(
uploaded_file, st.session_state.document_type
)
_det_type = uploaded_analysis.get("document_type", "document")
_conf = uploaded_analysis.get("confidence", "")
_conf_str = f" ยท {_conf} confidence" if _conf else ""
_upload_status.update(
label=f"โ
{_det_type} loaded{_conf_str}",
state="complete",
expanded=False,
)
else:
uploaded_analysis = parse_uploaded_file(
uploaded_file, st.session_state.document_type
)
# Replace stale data only when a genuinely new file is detected.
if _is_new_file:
# โโ Wipe ALL previous financial state โโโโโโโโโโโโโโโโโโโโโโโโ
for _k in (
"baseline", "metrics", "parsed_document",
"manual_revenue", "manual_expenses", "manual_profit",
"document_revenue", "document_expenses",
"generated_report", "analysis_report",
):
st.session_state.pop(_k, None)
for _k in [k for k in st.session_state if k.startswith("forecast_summary_")]:
del st.session_state[_k]
st.session_state["forecast_chat"] = []
# โโ Load fresh data from the new document โโโโโโโโโโโโโโโโโโโโ
st.session_state["parsed_document"] = uploaded_analysis
doc_rev = float(uploaded_analysis.get("revenue") or 0)
doc_exp = float(uploaded_analysis.get("expenses") or 0)
if doc_rev > 0 or doc_exp > 0:
st.session_state["manual_revenue"] = doc_rev
st.session_state["manual_expenses"] = doc_exp
new_baseline = {
"monthly_revenue": doc_rev,
"monthly_expenses": doc_exp,
"monthly_profit": doc_rev - doc_exp,
"months_of_data": int(uploaded_analysis.get("months_of_data") or 1),
"source": (
f"Uploaded {uploaded_analysis.get('document_type', 'document')}: "
f"{uploaded_file.name}"
),
"expenses_breakdown": uploaded_analysis.get("expenses_breakdown", {}),
}
st.session_state["baseline"] = new_baseline
st.session_state["metrics"] = compute_metrics(new_baseline)
st.session_state["_doc_fingerprint"] = _doc_fp
st.markdown('', unsafe_allow_html=True)
render_file_preview(uploaded_analysis)
if st.session_state.get("analysis_report"):
st.info("Document updated. Click **Re-run Analysis with Current Data** below to refresh all agent insights.")
else:
st.info("Document extracted successfully. Next step: click **Run Full Business Analysis** below.")
if _is_new_file:
components.html(
"""
""",
height=0,
width=0,
)
except Exception as exc:
upload_error = f"Could not parse the uploaded file. {exc}"
st.error(upload_error)
with st.expander("Need a sample template?", expanded=False):
st.markdown('Use one of these starter files if you want a clean format to upload.
', unsafe_allow_html=True)
render_download_buttons()
else:
st.markdown(
''
''
'โ Demo Mode โ no real accounts connected
',
unsafe_allow_html=True,
)
st.markdown('Connect a demo account
', unsafe_allow_html=True)
st.markdown('This is a safe demo flow. Pick a provider to preview how synced account data would improve the analysis.
', unsafe_allow_html=True)
render_clickable_provider_cards()
connected_preview = get_connected_demo(st.session_state.connected_provider)
if connected_preview:
render_connected_preview(connected_preview)
# โโ Analyze Button (bottom โ after optional data sources) โโโโโโโโโโโโโโโโโ
st.divider()
st.markdown('', unsafe_allow_html=True)
has_existing_analysis = bool(st.session_state.get("analysis_report"))
analyze_clicked = False
if not has_existing_analysis:
analyze_clicked = st.button(
"โก Run Full Business Analysis",
key="analyze_main",
type="primary",
use_container_width=True,
)
st.markdown(
'Runs all agents (cash flow, advisor, loan readiness, market, and summary).
',
unsafe_allow_html=True,
)
# Auto-trigger when demo flow set the flag (judge-friendly one-click path).
if not analyze_clicked and st.session_state.pop("auto_run_demo", False):
analyze_clicked = True
st.info("โจ Demo data loaded โ running full analysis automaticallyโฆ")
# Allow downstream analyze branch to use the previously parsed sample
# document when no fresh st.file_uploader event has occurred.
if uploaded_analysis is None and st.session_state.get("parsed_document"):
uploaded_analysis = st.session_state.get("parsed_document")
if analyze_clicked:
# Immediately scroll to the status panel so user sees agents working
components.html(
"""
""",
height=0,
width=0,
)
connected_analysis = get_connected_demo(st.session_state.connected_provider)
manual_figures = {
"revenue": float(st.session_state.manual_revenue),
"expenses": float(st.session_state.manual_expenses),
"debt": float(st.session_state.manual_debt),
"staff": int(st.session_state.manual_staff),
}
has_manual_figures = any(float(value or 0) > 0 for value in manual_figures.values())
business_profile = {
"business_type": st.session_state.business_type,
"location": st.session_state.location,
"products_services": st.session_state.products_services,
"main_question": st.session_state.main_question,
}
if (
not any(value.strip() for value in business_profile.values())
and not st.session_state.manual_notes.strip()
and not has_manual_figures
and uploaded_analysis is None
and connected_analysis is None
):
st.warning("Please provide business context, upload a document, connect demo transaction data, or combine them before running the analysis.")
elif upload_error and not st.session_state.manual_notes.strip() and not has_manual_figures and connected_analysis is None:
st.error("The uploaded file could not be analyzed, so there is not enough data to continue.")
else:
analysis_context, combined_parsed_data, basis_notes, source_labels, data_sources = combine_analysis_inputs(
business_profile,
st.session_state.manual_notes,
uploaded_analysis,
connected_analysis,
manual_figures=manual_figures,
)
with st.status("๐ง Duka AI โ analyzing your businessโฆ", expanded=True) as status:
pre_steps = sum([
1 if any(value.strip() for value in business_profile.values()) else 0,
1 if st.session_state.manual_notes.strip() else 0,
1 if has_manual_figures else 0,
1 if uploaded_analysis else 0,
1 if connected_analysis else 0,
])
total_steps = pre_steps + 5 # 5 downstream agents incl. executive summary
step_index = 0
progress = st.progress(0)
def _step(msg: str) -> None:
nonlocal step_index
step_index += 1
st.write(f"{step_index}/{total_steps} โข {msg}")
progress.progress(min(step_index / total_steps, 1.0))
if any(value.strip() for value in business_profile.values()):
_step("๐ Reading your business context and profileโฆ")
if st.session_state.manual_notes.strip():
_step("๐ Parsing your written business descriptionโฆ")
if has_manual_figures:
_step("๐ Locking in the numbers you enteredโฆ")
if uploaded_analysis:
_step(f"๐ Interpreting uploaded {uploaded_analysis.get('document_type', 'document').lower()}โฆ")
if connected_analysis:
_step(f"๐ Reading {connected_analysis.get('provider_selected', 'connected')} transaction dataโฆ")
completed_agents: set[str] = set()
def _on_agent_progress(agent_key: str) -> None:
rev = combined_parsed_data.get("revenue", 0) or 0
exp = combined_parsed_data.get("expenses", 0) or 0
profit = round(rev - exp, 0)
dynamic_msgs: dict[str, str] = {
"cashflow": (
f"๐ Cash Flow Agent โ analyzing {format_currency(rev)} revenue "
f"and {format_currency(exp)} expenses."
),
"advisor": (
f"๐ก Business Advisor Agent โ building recommendations "
f"from {format_currency(profit)} monthly profit."
),
"loan": (
f"๐ฆ Loan Readiness Agent โ calculating safe borrowing capacity "
f"from {format_currency(profit)} profit and {format_currency(combined_parsed_data.get('debt', 0) or 0)} existing debt."
),
"market": (
"๐ Market Intelligence Agent โ checking Zambian market trends "
f"for {combined_parsed_data.get('business_type') or 'your business type'}."
),
"summary": (
"๐ Executive Summary Agent โ writing your personalized "
f"financial summary for {business_profile.get('location') or 'your business'}."
),
}
msg = dynamic_msgs.get(agent_key)
if msg:
if agent_key not in completed_agents:
completed_agents.add(agent_key)
_step(msg)
else:
icon, name, action = AGENT_STATUS_MESSAGES.get(agent_key, ("โ๏ธ", agent_key, "Running..."))
if agent_key not in completed_agents:
completed_agents.add(agent_key)
_step(f"{icon} {name} โ {action}")
report = generate_business_report(
analysis_context,
parsed_data_override=combined_parsed_data,
analysis_basis=basis_notes,
source_labels=source_labels,
transaction_summary=connected_analysis,
business_profile=business_profile,
document_analysis=uploaded_analysis,
data_sources=data_sources,
progress_callback=_on_agent_progress,
)
progress.progress(1.0)
status.update(label="โ
Analysis complete! Scroll down to chat with your results.", state="complete", expanded=False)
st.session_state.analysis_report = report
st.session_state.pop("pending_demo_analysis_run", None)
_baseline = get_baseline(st.session_state)
st.session_state["baseline"] = _baseline
st.session_state["metrics"] = compute_metrics(_baseline) if _baseline else None
st.session_state.messages = [
{
"role": "assistant",
"content": report["final_summary"],
"type": "initial_analysis",
"report_key": True,
}
]
st.session_state["scroll_to_results"] = True
st.rerun()
# โโ Conversational Chat Interface โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ
report = st.session_state.get("analysis_report")
messages: list[dict] = st.session_state.get("messages", [])
scroll_requested = st.session_state.get("scroll_target") == "duka-ai-analyze-anchor"
inject_ui_behavior(
sample_cases=sample_cases,
selected_example_prompt=st.session_state.get("selected_example_prompt", ""),
selected_example_text=st.session_state.get("selected_example_prompt_text", ""),
selected_numbers_mode=st.session_state.get("numbers_entry_mode", "upload"),
scroll_requested=scroll_requested,
)
if scroll_requested:
st.session_state.scroll_target = None
if not messages:
st.markdown(
'Your chat workspace opens here after the first analysis. Fill in the form above, run the analysis, then ask follow-up questions about cash flow, borrowing, or growth.
',
unsafe_allow_html=True,
)
return
# โโ Auto-scroll to results after analysis โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ
st.markdown('', unsafe_allow_html=True)
if st.session_state.pop("scroll_to_results", False):
components.html(
"""
""",
height=0,
width=0,
)
stage_col, reset_col = st.columns([4, 1], gap="medium")
with stage_col:
st.markdown(
"""
""",
unsafe_allow_html=True,
)
with reset_col:
if st.button("โบ Reset / Start New Analysis", key="reset_analysis", use_container_width=True, help="Clear this analysis and start fresh"):
for _k in ["analysis_report", "baseline", "metrics"]:
st.session_state[_k] = None
st.session_state.messages = []
st.session_state.manual_notes = ""
st.session_state.manual_revenue = 0.0
st.session_state.manual_expenses = 0.0
st.session_state.manual_debt = 0.0
st.session_state.manual_staff = 0
st.session_state.business_type = ""
st.session_state.location = ""
st.session_state.products_services = ""
st.session_state.main_question = ""
st.session_state.selected_example_prompt = ""
st.session_state.selected_example_prompt_text = ""
st.session_state.hide_example_prompts = False
st.session_state.show_welcome = True
st.rerun()
st.markdown('', unsafe_allow_html=True)
# Render all chat messages
for msg in messages:
if msg["role"] == "user":
# Right-aligned bubble โ no Streamlit chat component needed
escaped = msg["content"].replace("&", "&").replace("<", "<").replace(">", ">")
st.markdown(
f'',
unsafe_allow_html=True,
)
else:
agent_key = msg.get("agent", "")
avatar_icon = AGENT_LABELS.get(agent_key, ("โจ", ""))[0] if agent_key else "โจ"
with st.chat_message("assistant", avatar=avatar_icon):
if msg.get("type") == "initial_analysis" and report:
render_initial_analysis_in_chat(report)
else:
if agent_key and agent_key in AGENT_LABELS:
icon, label = AGENT_LABELS[agent_key]
st.markdown(
f'{icon} {label}
',
unsafe_allow_html=True,
)
st.markdown(msg.get("content", ""), unsafe_allow_html=True)
if msg.get("visual"):
render_followup_visual(msg["visual"])
if msg.get("chart"):
render_followup_chart(msg["chart"])
# โโ Suggestion pills โ above input โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ
if report and len(messages) == 1:
st.markdown('', unsafe_allow_html=True)
sug_cols = st.columns(3, gap="small")
for sug_i, (suggestion_label, suggestion_prompt) in enumerate(FOLLOWUP_SUGGESTION_PILLS):
with sug_cols[sug_i % 3]:
if st.button(suggestion_label, key=f"sug_{sug_i}", use_container_width=True):
agents_to_run = _resolve_agents(suggestion_prompt, messages)
primary_agent = agents_to_run[0]
st.session_state.messages.append({"role": "user", "content": suggestion_prompt})
with st.spinner(f"๐ฌ {suggestion_label} โฆ"):
result = answer_followup_question(suggestion_prompt, report, history=None)
visual = build_followup_visual(suggestion_prompt, report)
chart = build_chart_data(suggestion_prompt, report, st.session_state.messages)
followup_msg: dict[str, Any] = {
"role": "assistant",
"content": _normalize_assistant_response(result["answer"]),
"agent": primary_agent,
}
if visual:
followup_msg["visual"] = visual
if chart:
followup_msg["chart"] = chart
st.session_state.messages.append(followup_msg)
st.session_state.last_agent = primary_agent
st.session_state.scroll_to_chat_conversation = True
st.rerun()
# โโ Chat input with streaming โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ
if report:
st.markdown('', unsafe_allow_html=True)
_injected = st.session_state.pop("pending_chat_question", None)
if prompt := (_injected or st.chat_input("Ask about your business...")):
agents_to_run = _resolve_agents(prompt, messages)
primary_agent = agents_to_run[0]
icon, label = AGENT_LABELS.get(primary_agent, ("๐ก", "Business Advisor"))
# Right-aligned user bubble (consistent with message loop)
escaped_prompt = prompt.replace("&", "&").replace("<", "<").replace(">", ">")
st.markdown(
f'',
unsafe_allow_html=True,
)
st.session_state.messages.append({"role": "user", "content": prompt})
# ChatGPT-style anchoring: pin the user's question at the top of
# the viewport so the streaming response fills the screen below it.
# Hammered instant jumps in the first ~1.3s overpower Streamlit's
# scroll-restore on rerun. After that, the script BOWS OUT and lets
# the user scroll freely โ never yanks them back up.
components.html(
"""
""",
height=0,
width=0,
)
with st.chat_message("assistant", avatar=icon):
st.markdown(
f'{icon} {label}
',
unsafe_allow_html=True,
)
if settings.llm_enabled:
# ChatGPT-style: stream tokens into the UI with typewriter effect + auto-follow scroll
stream_gen = stream_followup_for_agent(
primary_agent,
prompt,
report,
st.session_state.messages,
)
try:
full_response = st.write_stream(stream_gen, cursor="โ")
except TypeError:
# Older Streamlit versions don't support `cursor` kwarg
full_response = st.write_stream(stream_gen)
if isinstance(full_response, list):
full_response = "".join(str(x) for x in full_response)
else:
full_response = str(full_response or "")
else:
result = answer_followup_question(prompt, report, history=None)
full_response = result["answer"]
st.markdown(full_response)
visual = build_followup_visual(prompt, report)
if visual:
render_followup_visual(visual)
chart = build_chart_data(prompt, report, st.session_state.messages)
if chart:
render_followup_chart(chart)
followup_msg = {
"role": "assistant",
"content": _normalize_assistant_response(str(full_response)),
"agent": primary_agent,
}
if visual:
followup_msg["visual"] = visual
if chart:
followup_msg["chart"] = chart
st.session_state.messages.append(followup_msg)
st.session_state.last_agent = primary_agent
st.session_state.scroll_to_chat_conversation = True
# Anchor at bottom of chat thread (kept for compatibility with other flows
# that may toggle scroll_to_chat_conversation).
st.markdown('', unsafe_allow_html=True)
if st.session_state.pop("scroll_to_chat_conversation", False):
# Pin the latest user question to the top of viewport (ChatGPT-style)
# so the response below it is visible. Bows out as soon as the user
# shows scroll intent.
components.html(
"""
""",
height=0,
width=0,
)
if __name__ == "__main__":
main()