diff --git a/README.md b/README.md index 26cb9cc..84bf485 100644 --- a/README.md +++ b/README.md @@ -26,11 +26,33 @@ If you plan to run the app in a conda or virtual environment, make sure to set u 3. Run the application with the following command: ```bash - streamlit run recommender.py + streamlit run main.py ``` 4. Once running, the app will be accessible at `localhost:8501`. +### Personal Finance Domain Demo + +This repository also includes a personal finance domain adaptation that recommends budgeting and money-habit actions from a local catalog. It can run with AO packages installed, and it also includes a deterministic fallback path so reviewers can test the domain without private packages or paid APIs. + +Run the finance demo: + +```bash +streamlit run finance_recommender.py +``` + +Run the fallback CLI: + +```bash +python finance_recommender.py +``` + +Run the finance domain tests: + +```bash +python -m unittest tests/test_finance_domain.py +``` + ### Docker Installation @@ -55,10 +77,11 @@ You're done! Access the app at `localhost:8501` in your browser. The recommender system works by loading a set of random video links. Once the user hits the Run button, a video will be shown, and the system will suggest whether it recommends the video or not. The user can then provide feedback using "pain" or "pleasure" signals to guide the recommendation process. Based on this feedback, the system adjusts its responses and suggests another video. This cycle continues, allowing for more accurate and personalized recommendations over time. +The finance demo follows the same continuous-feedback pattern with a different domain. It encodes each financial action into the same eight-bit AO-compatible input shape using action focus, time horizon, risk, and the user's current money goal. User feedback updates fallback rankings immediately and trains an AO Agent when optional AO packages are available. + ## Contributing Fork the repository, make your changes, and submit a pull request for review. - diff --git a/arch__FinanceRecommender.py b/arch__FinanceRecommender.py new file mode 100644 index 0000000..68f62d1 --- /dev/null +++ b/arch__FinanceRecommender.py @@ -0,0 +1,15 @@ +# -*- coding: utf-8 -*- +"""AO architecture for the personal finance recommender demo.""" + +import ao_arch as ar + + +description = "Personal Finance Recommender" + +# focus, horizon, risk, user goal +arch_i = [3, 2, 1, 2] +arch_z = [10] +arch_c = [] +connector_function = "full_conn" + +arch = ar.Arch(arch_i, arch_z, arch_c, connector_function, description) diff --git a/finance_domain.py b/finance_domain.py new file mode 100644 index 0000000..ae0676c --- /dev/null +++ b/finance_domain.py @@ -0,0 +1,244 @@ +"""Personal finance recommendation domain for the AO recommender demo.""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import Iterable + + +FOCUS_BITS = { + "savings": [0, 0, 0], + "debt": [0, 0, 1], + "investing": [0, 1, 0], + "cash-flow": [0, 1, 1], + "protection": [1, 0, 0], +} + +HORIZON_BITS = { + "now": [0, 0], + "short": [0, 1], + "long": [1, 1], +} + +GOAL_BITS = { + "build-buffer": [0, 0], + "reduce-stress": [0, 1], + "grow-wealth": [1, 0], + "simplify": [1, 1], +} + + +@dataclass(frozen=True) +class FinanceAction: + """A financial habit or action that can be recommended.""" + + name: str + focus: str + horizon: str + effort: str + risk: str + monthly_cash_impact: int + contexts: tuple[str, ...] + description: str + + +FINANCE_ACTIONS: tuple[FinanceAction, ...] = ( + FinanceAction( + name="Start a starter emergency fund", + focus="savings", + horizon="short", + effort="medium", + risk="low", + monthly_cash_impact=150, + contexts=("build-buffer", "reduce-stress"), + description="Automatically move a small fixed amount into a separate emergency account.", + ), + FinanceAction( + name="Audit recurring subscriptions", + focus="cash-flow", + horizon="now", + effort="low", + risk="low", + monthly_cash_impact=45, + contexts=("simplify", "reduce-stress"), + description="Cancel or downgrade subscriptions that no longer support current priorities.", + ), + FinanceAction( + name="Snowball the smallest debt", + focus="debt", + horizon="short", + effort="medium", + risk="low", + monthly_cash_impact=100, + contexts=("reduce-stress", "simplify"), + description="Pay minimums on every debt and send extra cash to the smallest balance.", + ), + FinanceAction( + name="Refinance high-interest debt", + focus="debt", + horizon="short", + effort="high", + risk="medium", + monthly_cash_impact=120, + contexts=("reduce-stress", "build-buffer"), + description="Compare lower-rate options before interest costs crowd out other goals.", + ), + FinanceAction( + name="Increase retirement contribution by one percent", + focus="investing", + horizon="long", + effort="low", + risk="medium", + monthly_cash_impact=-60, + contexts=("grow-wealth", "simplify"), + description="Make a small automatic retirement increase that compounds quietly over time.", + ), + FinanceAction( + name="Build a low-cost index fund habit", + focus="investing", + horizon="long", + effort="medium", + risk="medium", + monthly_cash_impact=-100, + contexts=("grow-wealth", "build-buffer"), + description="Use recurring purchases into diversified, low-fee index funds.", + ), + FinanceAction( + name="Create a weekly cash-flow checkpoint", + focus="cash-flow", + horizon="now", + effort="low", + risk="low", + monthly_cash_impact=80, + contexts=("simplify", "reduce-stress"), + description="Review upcoming bills, income, and spending once a week.", + ), + FinanceAction( + name="Negotiate bills and insurance renewals", + focus="cash-flow", + horizon="short", + effort="medium", + risk="low", + monthly_cash_impact=70, + contexts=("simplify", "build-buffer"), + description="Ask providers for retention discounts or compare quotes before renewals.", + ), + FinanceAction( + name="Set a one-month bill buffer", + focus="protection", + horizon="long", + effort="high", + risk="low", + monthly_cash_impact=250, + contexts=("build-buffer", "reduce-stress"), + description="Accumulate enough cash so next month's fixed bills are already covered.", + ), + FinanceAction( + name="Review insurance deductibles", + focus="protection", + horizon="short", + effort="medium", + risk="medium", + monthly_cash_impact=35, + contexts=("reduce-stress", "build-buffer"), + description="Match deductibles to cash reserves so premiums and risk stay balanced.", + ), + FinanceAction( + name="Create a values-based spending rule", + focus="cash-flow", + horizon="now", + effort="low", + risk="low", + monthly_cash_impact=60, + contexts=("simplify", "grow-wealth"), + description="Define what spending is always worth it and what gets paused by default.", + ), + FinanceAction( + name="Start a side-income experiment", + focus="savings", + horizon="short", + effort="high", + risk="medium", + monthly_cash_impact=200, + contexts=("grow-wealth", "build-buffer"), + description="Run a small, time-boxed income experiment before committing heavily.", + ), +) + + +def encode_finance_action(action: FinanceAction, goal: str = "build-buffer") -> list[int]: + """Encode an action plus user goal into the AO eight-bit input shape.""" + + if action.focus not in FOCUS_BITS: + raise ValueError(f"Unknown focus: {action.focus}") + if action.horizon not in HORIZON_BITS: + raise ValueError(f"Unknown horizon: {action.horizon}") + if goal not in GOAL_BITS: + raise ValueError(f"Unknown goal: {goal}") + + risk_bit = [1 if action.risk == "medium" else 0] + return FOCUS_BITS[action.focus] + HORIZON_BITS[action.horizon] + risk_bit + GOAL_BITS[goal] + + +def score_finance_action( + action: FinanceAction, + goal: str, + available_cash: int, + feedback: dict[str, int] | None = None, +) -> int: + """Score an action using deterministic preferences and optional feedback.""" + + score = 0 + + if goal in action.contexts: + score += 35 + + if action.monthly_cash_impact > 0: + score += min(action.monthly_cash_impact // 10, 25) + elif available_cash >= abs(action.monthly_cash_impact): + score += 15 + else: + score -= 20 + + if action.effort == "low": + score += 15 + elif action.effort == "medium": + score += 8 + + if goal == "grow-wealth" and action.focus == "investing": + score += 20 + if goal == "reduce-stress" and action.risk == "low": + score += 10 + + if feedback: + score += feedback.get(action.name, 0) * 12 + + return score + + +def recommend_finance_actions( + goal: str = "build-buffer", + available_cash: int = 100, + feedback: dict[str, int] | None = None, + limit: int = 3, + actions: Iterable[FinanceAction] = FINANCE_ACTIONS, +) -> list[tuple[FinanceAction, int]]: + """Return the highest-scoring finance actions for the user's current context.""" + + if goal not in GOAL_BITS: + raise ValueError(f"Unknown goal: {goal}") + + ranked = [ + (action, score_finance_action(action, goal, available_cash, feedback)) + for action in actions + ] + ranked.sort(key=lambda item: (item[1], item[0].monthly_cash_impact, item[0].name), reverse=True) + return ranked[:limit] + + +def apply_feedback(feedback: dict[str, int], action_name: str, liked: bool) -> dict[str, int]: + """Return updated feedback weights for a finance action.""" + + updated = dict(feedback) + updated[action_name] = updated.get(action_name, 0) + (1 if liked else -1) + return updated diff --git a/finance_recommender.py b/finance_recommender.py new file mode 100644 index 0000000..2f51685 --- /dev/null +++ b/finance_recommender.py @@ -0,0 +1,156 @@ +"""Streamlit and CLI demo for a personal finance recommender domain.""" + +from __future__ import annotations + +from finance_domain import ( + FINANCE_ACTIONS, + apply_feedback, + encode_finance_action, + recommend_finance_actions, +) + + +def create_agent(): + """Create an AO Agent when optional AO packages are installed.""" + + try: + import ao_core as ao + from arch__FinanceRecommender import arch + except Exception: + return None + + agent = ao.Agent(arch, notes="Personal Finance Agent") + for _ in range(4): + agent.reset_state() + agent.reset_state(training=True) + return agent + + +def ao_percentage(agent, binary_input: list[int]) -> int | None: + """Return an AO recommendation percentage, or None when AO is unavailable.""" + + if agent is None: + return None + + agent.reset_state() + response = None + for _ in range(5): + response = agent.next_state(INPUT=binary_input, print_result=False) + + if response is None: + return None + + return round(sum(1 for value in response if value == 1) / len(response) * 100) + + +def train_agent(agent, binary_input: list[int], liked: bool) -> None: + """Train the optional AO Agent on user feedback.""" + + if agent is None: + return + + import numpy as np + + label_value = 1 if liked else 0 + label = np.full(agent.arch.Z__flat.shape, label_value, dtype=np.int8) + for _ in range(5 if liked else 10): + agent.reset_state() + agent.next_state(INPUT=binary_input, LABEL=label, print_result=False, unsequenced=True) + + +def run_cli() -> None: + """Print fallback recommendations without requiring Streamlit or AO packages.""" + + print("Top personal finance recommendations:") + for action, score in recommend_finance_actions(goal="build-buffer", available_cash=150): + print(f"- {action.name} ({score})") + print(f" input={encode_finance_action(action, 'build-buffer')}") + print(f" {action.description}") + + +def run_streamlit() -> None: + """Run the interactive Streamlit demo.""" + + import streamlit as st + + st.set_page_config( + page_title="Personal Finance Recommender by AO Labs", + page_icon="misc/ao_favicon.png", + layout="wide", + initial_sidebar_state="expanded", + ) + + if "finance_feedback" not in st.session_state: + st.session_state.finance_feedback = {} + if "finance_agent" not in st.session_state: + st.session_state.finance_agent = create_agent() + + st.title("Personal Finance Recommender") + st.write("A domain adaptation of the AO recommender for budgeting and money habits.") + + with st.sidebar: + goal = st.selectbox( + "Current money goal", + ("build-buffer", "reduce-stress", "grow-wealth", "simplify"), + format_func=lambda value: value.replace("-", " ").title(), + ) + available_cash = st.slider("Monthly cash available for a new habit", 0, 500, 150, 25) + ao_status = "available" if st.session_state.finance_agent is not None else "fallback mode" + st.write(f"AO Agent: {ao_status}") + + ranked = recommend_finance_actions( + goal=goal, + available_cash=available_cash, + feedback=st.session_state.finance_feedback, + limit=len(FINANCE_ACTIONS), + ) + + for action, fallback_score in ranked[:5]: + binary_input = encode_finance_action(action, goal) + ao_score = ao_percentage(st.session_state.finance_agent, binary_input) + display_score = ao_score if ao_score is not None else fallback_score + + st.subheader(action.name) + st.write(action.description) + st.write( + { + "focus": action.focus, + "horizon": action.horizon, + "risk": action.risk, + "monthly_cash_impact": action.monthly_cash_impact, + "encoded_input": binary_input, + "score": display_score, + } + ) + + left, right = st.columns(2) + if left.button("Recommend more like this", key=f"like-{action.name}"): + st.session_state.finance_feedback = apply_feedback( + st.session_state.finance_feedback, action.name, liked=True + ) + train_agent(st.session_state.finance_agent, binary_input, liked=True) + st.rerun() + if right.button("Recommend less like this", key=f"less-{action.name}"): + st.session_state.finance_feedback = apply_feedback( + st.session_state.finance_feedback, action.name, liked=False + ) + train_agent(st.session_state.finance_agent, binary_input, liked=False) + st.rerun() + + +def is_streamlit_runtime() -> bool: + """Detect whether this script is being executed by Streamlit.""" + + try: + from streamlit.runtime.scriptrunner import get_script_run_ctx + except Exception: + return False + + return get_script_run_ctx() is not None + + +if __name__ == "__main__": + if is_streamlit_runtime(): + run_streamlit() + else: + run_cli() diff --git a/tests/test_finance_domain.py b/tests/test_finance_domain.py new file mode 100644 index 0000000..b3199e2 --- /dev/null +++ b/tests/test_finance_domain.py @@ -0,0 +1,53 @@ +import unittest + +from finance_domain import ( + FINANCE_ACTIONS, + apply_feedback, + encode_finance_action, + recommend_finance_actions, +) + + +class FinanceDomainTests(unittest.TestCase): + def test_encoding_matches_ao_input_shape(self): + action = FINANCE_ACTIONS[0] + + encoded = encode_finance_action(action, goal="build-buffer") + + self.assertEqual(len(encoded), 8) + self.assertTrue(all(bit in (0, 1) for bit in encoded)) + + def test_recommendations_prioritize_matching_goal(self): + ranked = recommend_finance_actions(goal="grow-wealth", available_cash=200, limit=3) + + self.assertTrue(any(action.focus == "investing" for action, _ in ranked)) + + def test_low_cash_penalizes_cash_negative_actions(self): + low_cash = recommend_finance_actions(goal="grow-wealth", available_cash=0, limit=5) + + self.assertTrue(all(action.monthly_cash_impact >= -100 for action, _ in low_cash)) + + def test_feedback_changes_ranking(self): + baseline = recommend_finance_actions(goal="simplify", available_cash=100, limit=1)[0][0] + feedback = apply_feedback({}, "Start a side-income experiment", liked=True) + feedback = apply_feedback(feedback, "Start a side-income experiment", liked=True) + feedback = apply_feedback(feedback, "Start a side-income experiment", liked=True) + feedback = apply_feedback(feedback, "Start a side-income experiment", liked=True) + + updated = recommend_finance_actions( + goal="simplify", + available_cash=100, + feedback=feedback, + limit=1, + )[0][0] + + self.assertNotEqual(baseline.name, updated.name) + self.assertEqual(updated.name, "Start a side-income experiment") + + def test_unknown_goal_is_rejected(self): + with self.assertRaises(ValueError): + recommend_finance_actions(goal="unknown") + + +if __name__ == "__main__": + unittest.main()