feat(simulation): simulate your portfolio's future (#136)

* feat(simulation): create timeline structure, display worth evolution

* refactor: split simulator classes in separate files

* fix: ask before fetching from N26

* feat: add default event to apply yearly performance

* feat: add autobalance default quarterly event

* docs: add docstrings

Author: MadeInPierre

README.md

@@ -55,9 +55,9 @@ Statistics and visualizations will be added soon!
 ## ✨ Features
 
 1. **✅ Portfolio:** Organize your assets, set targets, and sync with your Finary account.
-2. **⏳ Web dashboard:** Generate global statistics and graphs to understand each line and folder.
-3. **⏳ Assistant:** Get monthly recommendations on where to invest next to meet your goals.
-4. **🔜 Simulator:** Define your life goals and events, simulate your portfolio's future.
+2. **✅ Assistant:** Get monthly recommendations on where to invest next to meet your goals.
+3. **✅ Simulator:** Define your life goals and events, simulate your portfolio's future.
+4. **⏳ Web dashboard:** Generate global statistics and graphs to understand each line and folder.
 5. **🙏 Extensions:** Make this tool work for other people's situations, contributions needed 👀
 
 You can check the [current development status](https://github.com/users/MadeInPierre/projects/4). Contributions are warmly welcome!

finalynx/__init__.py

@@ -29,7 +29,11 @@
 from .dashboard import Dashboard
 
 # Simulator
-from .simulator import Simulator
+from .simulator import Simulator, Simulation, Timeline
+from .simulator import Action, AddLineAmount, SetLineAmount
+from .simulator import Event, Salary
+from .simulator import DeltaRecurrence, MonthlyRecurrence
+from datetime import date
 
 # Main
 from .assistant import Assistant

finalynx/__main__.py

@@ -3,11 +3,11 @@
 
 Currently, Finalynx does not support direct calls and only prints the command-line usage description to the console.
 """
-from docopt import docopt
+from docopt import docopt  # type: ignore[import]
 
 from .__meta__ import __version__
 from .parse.json import ImportJSON
-from .usage import __doc__
+from .usage import __doc__  # type: ignore[import]
 
 if __name__ == "__main__":
     args = docopt(__doc__, version=__version__)

finalynx/assistant.py

@@ -2,13 +2,13 @@
 import os
 from datetime import date
 from typing import Any
+from typing import Dict
 from typing import List
 from typing import Optional
 from typing import Tuple
-from typing import TYPE_CHECKING
 
 import finalynx.theme
-from docopt import docopt
+from docopt import docopt  # type: ignore[import]
 from finalynx import Dashboard
 from finalynx import Fetch
 from finalynx import Portfolio
@@ -21,21 +21,20 @@
 from finalynx.portfolio.bucket import Bucket
 from finalynx.portfolio.envelope import Envelope
 from finalynx.portfolio.folder import Sidecar
-from html2image import Html2Image
+from finalynx.simulator.timeline import Simulation
+from finalynx.simulator.timeline import Timeline
+from html2image import Html2Image  # type: ignore[import]
 from rich import inspect  # noqa F401
 from rich import pretty
 from rich import print  # noqa F401
 from rich import traceback
 from rich.columns import Columns
 from rich.console import Console
 from rich.panel import Panel
+from rich.prompt import Confirm
 from rich.text import Text
 from rich.tree import Tree
 
-
-if TYPE_CHECKING:
-    from rich.console import ConsoleRenderable
-
 from .__meta__ import __version__
 from .console import console
 from .usage import __doc__
@@ -65,9 +64,11 @@ class Assistant:
 
     def __init__(
         self,
+        # Main structure elements
         portfolio: Portfolio,
         buckets: Optional[List[Bucket]] = None,
         envelopes: Optional[List[Envelope]] = None,
+        # Portfolio options
         ignore_orphans: bool = False,
         clear_cache: bool = False,
         force_signin: bool = False,
@@ -81,15 +82,18 @@ def __init__(
         active_sources: Optional[List[str]] = None,
         theme: Optional[finalynx.theme.Theme] = None,
         sidecars: Optional[List[Sidecar]] = None,
+        ignore_argv: bool = False,
+        # Budget options
         check_budget: bool = False,
         interactive: bool = False,
-        ignore_argv: bool = False,
+        # Simulation options
+        simulation: Optional[Simulation] = None,
     ):
         self.portfolio = portfolio
         self.buckets = buckets if buckets else []
         self.envelopes = envelopes if envelopes else []
 
-        # Options that can either be set in the constructor or from the command line options, type --help
+        # Options that can either be set in the constructor or from the command line options, see --help
         self.ignore_orphans = ignore_orphans
         self.clear_cache = clear_cache
         self.force_signin = force_signin
@@ -104,6 +108,7 @@ def __init__(
         self.sidecars = sidecars if sidecars else []
         self.check_budget = check_budget
         self.interactive = interactive
+        self.simulation = simulation
 
         # Set the global color theme if specified
         if theme:
@@ -117,6 +122,9 @@ def __init__(
         self._fetch = Fetch(self.portfolio, self.clear_cache, self.ignore_orphans)
         self.budget = Budget()
 
+        # Initialize the simulation timeline with the initial user events
+        self._timeline = Timeline(simulation, self.portfolio, self.buckets) if simulation else None
+
     def add_source(self, source: SourceBaseLine) -> None:
         """Register a source, either defined in your own config or from the available Finalynx sources
         using `from finalynx.fetch.source_any import SourceAny`."""
@@ -204,8 +212,8 @@ def run(self) -> None:
 
         # Render the console elements
         main_frame = self.render_mainframe()
-        panels = self.render_panels()
-        renders: List[Any] = [main_frame, panels]
+        dict_panels, render_panels = self.render_panels()
+        renders: List[Any] = [main_frame, render_panels]
         if self.check_budget:
             renders.append(self.budget.render_expenses())
 
@@ -217,9 +225,32 @@ def run(self) -> None:
         if self.show_data:
             console.print(Panel(fetched_tree, title="Fetched data"))
 
+        # Run the simulation if there are events defined
+        if self._timeline and not self._timeline.is_finished:
+            console.log(f"Running simulation until {self._timeline.end_date}...")
+            tree = Tree("\n[bold]Worth", guide_style=TH().TREE_BRANCH)
+
+            def append_worth(year: int, amount: float) -> None:
+                tree.add(f"[{TH().TEXT}]{year}:       [{TH().ACCENT}][bold]{round(amount / 1000):>4}[/] k€")
+
+            append_worth(date.today().year, self.portfolio.get_amount())
+            for year in range(date.today().year + 5, self._timeline.end_date.year, 5):
+                self._timeline.goto(date(year, 1, 1))
+                append_worth(year, self.portfolio.get_amount())
+            self._timeline.run()
+            append_worth(self._timeline.current_date.year, self.portfolio.get_amount())
+            dict_panels["performance"].add(tree)
+
+            console.log(f"    Portfolio will be worth [{TH().ACCENT}]{self.portfolio.get_amount():.0f} €[/]")
+
         # Display the entire portfolio and associated recommendations
         for render in renders:
             console.print("\n\n", render)
+        console.print("\n")
+
+        # TODO replace with a command option
+        if self._timeline and Confirm.ask(f"Display your future portfolio in {self._timeline.end_date}?"):
+            console.print("\n\n", self.render_mainframe())
 
         # Interactive review of the budget expenses if enabled
         if self.check_budget and self.interactive:
@@ -267,33 +298,37 @@ def render_mainframe(self) -> Columns:
 
         return Columns(main_frame, padding=(0, 0))  # type: ignore
 
-    def render_panels(self) -> Columns:
+    def render_panels(self) -> Tuple[Dict[str, Any], Columns]:
         """Renders the default set of panels used in the default console view when calling run()."""
 
         def panel(title: str, content: Any) -> Panel:
             return Panel(content, title=title, padding=(1, 2), expand=False, border_style=TH().PANEL)
 
-        # Final set of results to be displayed
-        panels: List[ConsoleRenderable] = [
-            Text(" "),
-            panel("Recommendations", render_recommendations(self.portfolio, self.envelopes)),
-            panel("Performance", self.render_performance_report()),
-        ]
+        dict_items: Dict[str, Any] = {
+            "recommendations": render_recommendations(self.portfolio, self.envelopes),
+            "performance": self.render_performance_report(),
+        }
 
         # Add the budget panel if enabled
         if self.check_budget:
-            panels.append(panel("Budget", self.budget.render_summary()))
+            dict_items["budget"] = self.budget.render_summary()
 
-        return Columns(panels, padding=(2, 2))
+        return dict_items, Columns(
+            [Text(" ")] + [panel(k.capitalize(), v) for k, v in dict_items.items()], padding=(2, 2)
+        )
 
     def render_performance_report(self) -> Tree:
         """Print the current and ideal global expected performance. Call either run() or initialize() first."""
         perf = self.portfolio.get_perf(ideal=False).expected
         perf_ideal = self.portfolio.get_perf(ideal=True).expected
 
-        tree = Tree("Global Performance", hide_root=True)
-        tree.add(f"[{TH().TEXT}]Current:  [bold][{TH().ACCENT}]{perf:.1f} %[/] / year")
-        tree.add(f"[{TH().TEXT}]Planned:  [bold][{TH().ACCENT}]{perf_ideal:.1f} %[/] / year")
+        tree = Tree("Global Performance", hide_root=True, guide_style=TH().TREE_BRANCH)
+        node = tree.add("[bold]Yield")
+        node.add(f"[{TH().TEXT}]Current:    [bold][{TH().ACCENT}]{perf:.2f} %[/] / year")
+        node.add(f"[{TH().TEXT}]Planned:    [bold][{TH().ACCENT}]{perf_ideal:.2f} %[/] / year")
+
+        if self.simulation:
+            node.add(f"[{TH().TEXT}]Inflation:  [bold][gold1]{self.simulation.inflation:.2f} %[/] / year")
         return tree
 
     def dashboard(self) -> None:
@@ -351,7 +386,7 @@ def export_img(
         # Export the entire portfolio tree to HTML and set the zoom
         dashboard_console = Console(record=True, file=open(os.devnull, "w"))
         dashboard_console.print(self.render_mainframe())
-        dashboard_console.print(self.render_panels())
+        dashboard_console.print(self.render_panels()[1])
         output_html = dashboard_console.export_html().replace("body {", f"body {{\n    zoom: {zoom};")
 
         # Convert the HTML to PNG

finalynx/budget/budget.py

@@ -5,6 +5,7 @@
 from typing import Union
 
 import gspread
+from rich.prompt import Confirm
 from rich.table import Table
 from rich.tree import Tree
 
@@ -33,7 +34,7 @@ def __init__(self) -> None:
 
         # Initialize the list of expenses, will be fetched later
         self.expenses: List[Expense] = []
-        self.n_new_expenses: int = -1
+        self.n_new_expenses: int = 0
         self.balance: float = 0.0
 
         # Private copy that only includes expenses that need user review (calculated only once)
@@ -44,11 +45,6 @@ def fetch(self, clear_cache: bool, force_signin: bool = False) -> Tree:
         This method also updates the google sheets table with the newly found expenses and
         prepares the list of "pending" expenses that need user reviews."""
 
-        # Initialize the N26 client with the credentials
-        source = SourceN26(force_signin)
-        tree = source.fetch(clear_cache=bool(clear_cache or force_signin))
-        self.balance = source.balance
-
         # Connect to the Google Sheet that serves as the database of expenses
         with console.status(f"[bold {TH().ACCENT}]Connecting to Google Sheets...", spinner_style=TH().ACCENT):
             try:
@@ -67,22 +63,30 @@ def fetch(self, clear_cache: bool, force_signin: bool = False) -> Tree:
         # Fetch the latest values from the the sheet
         sheet_values = self._sheet.get_all_values()
 
-        # Get the new expenses from the source that are not in the sheet yet
-        last_timestamp = max([int(row[0]) for row in sheet_values if str(row[0]).isdigit()])
-        new_expenses = list(reversed([e for e in source.get_expenses() if e.timestamp > last_timestamp]))
-        self.n_new_expenses = len(new_expenses)
-
-        # Add the new expenses to the sheet
-        if self.n_new_expenses > 0:
-            first_empty_row = len(sheet_values) + 1
-
-            self._sheet.update(
-                f"A{first_empty_row}:D{first_empty_row + len(new_expenses)}",
-                [d.to_list()[:4] for d in new_expenses],
-            )
+        # Initialize the N26 client with the credentials
+        if Confirm.ask("Fetch expenses from N26?", default=True):
+            source = SourceN26(force_signin)
+            tree = source.fetch(clear_cache=bool(clear_cache or force_signin))
+            self.balance = source.balance
+
+            # Get the new expenses from the source that are not in the sheet yet
+            last_timestamp = max([int(row[0]) for row in sheet_values if str(row[0]).isdigit()])
+            new_expenses = list(reversed([e for e in source.get_expenses() if e.timestamp > last_timestamp]))
+            self.n_new_expenses = len(new_expenses)
+
+            # Add the new expenses to the sheet
+            if self.n_new_expenses > 0:
+                first_empty_row = len(sheet_values) + 1
+
+                self._sheet.update(
+                    f"A{first_empty_row}:D{first_empty_row + len(new_expenses)}",
+                    [d.to_list()[:4] for d in new_expenses],
+                )
 
-        # Fetch the latest values from the sheet again to get the complete list of expenses
-        sheet_values = self._fetch_sheet_values()  # TODO improve
+            # Fetch the latest values from the sheet again to get the complete list of expenses
+            sheet_values = self._fetch_sheet_values()  # TODO improve
+        else:
+            tree = Tree("N26 Skipped.")
 
         # From now on, we will work with the up-to-date list of expenses
         self.expenses = [Expense.from_list(line, i + 2) for i, line in enumerate(sheet_values[1:])]
@@ -97,14 +101,15 @@ def fetch(self, clear_cache: bool, force_signin: bool = False) -> Tree:
 
     def render_expenses(self) -> Union[Table, str]:
         # Make sure we are already connected to the source and the sheet
-        assert self.n_new_expenses > -1, "Call `fetch()` first"
         assert self._pending_expenses is not None, "Call `fetch()` first"
 
         # Display the table of pending expenses
         n_pending = len(self._pending_expenses)
 
         if n_pending == 0:
-            return f"[green]No pending expenses 🎉 [dim white]N26 Balance: {self.balance:.2f} €\n"
+            return "[green]No pending expenses 🎉" + (
+                f" [dim white]N26 Balance: {self.balance:.2f} €\n" if self.balance > 0.001 else "\n"
+            )
 
         return _render_expenses_table(
             self._pending_expenses[-Budget.MAX_DISPLAY_ROWS :],  # noqa: E203
@@ -119,7 +124,6 @@ def render_summary(self) -> Tree:
         """Render a summary of the budget, mainly the current and previous month's totals."""
 
         # Make sure we are already connected to the source and the sheet
-        assert self.n_new_expenses > -1, "Call `fetch()` first"
         assert self.expenses is not None, "Call `fetch()` first"
         tree = Tree("Budget", hide_root=True, guide_style=TH().HINT)