diff --git a/finalynx/analyzer/asset_class.py b/finalynx/analyzer/asset_class.py
index f15062d..6e287b6 100644
--- a/finalynx/analyzer/asset_class.py
+++ b/finalynx/analyzer/asset_class.py
@@ -1,3 +1,4 @@
+from datetime import date
from typing import Any
from typing import Dict
@@ -72,6 +73,31 @@ def analyze(self) -> Dict[str, Any]:
sum of investments corresponding to each class."""
return self._recursive_merge(self.node)
+ def analyzeTime(self, target_date: date) -> Dict[str, float]:
+ """:returns: A dictionary with keys as the asset class names and values as the
+ sum of investments corresponding to each class."""
+ return self._recursive_mergeTime(self.node, target_date)
+
+ def _recursive_mergeTime(self, node: Node, target_date: date) -> Dict[str, Any]:
+ """Internal method for recursive searching."""
+ total = {c.value: 0.0 for c in AssetClass}
+
+ # Lines simply return their own amount
+ if isinstance(node, Line):
+ total[node.asset_class.value] = node.get_amount()
+ return total
+
+ # Folders merge what the children return
+ elif isinstance(node, Folder):
+ for child in node.children:
+ for key, value in self._recursive_mergeTime(child, target_date).items():
+ total[key] += value
+ return total
+
+ # Safeguard for future versions
+ else:
+ raise ValueError(f"Unknown node type '{type(node)}'.")
+
def _recursive_merge(self, node: Node) -> Dict[str, Any]:
"""Internal method for recursive searching."""
result: Dict[str, Any] = {
diff --git a/finalynx/analyzer/envelopes.py b/finalynx/analyzer/envelopes.py
index 7486b17..988ea00 100644
--- a/finalynx/analyzer/envelopes.py
+++ b/finalynx/analyzer/envelopes.py
@@ -1,6 +1,9 @@
+from datetime import date
from typing import Any
from typing import Dict
+from finalynx.portfolio.constants import EnvelopeClass
+
from ..portfolio import Folder
from ..portfolio import Line
from ..portfolio import Node
@@ -19,6 +22,37 @@ def analyze(self) -> Dict[str, float]:
sum of investments corresponding to each class."""
return self._recursive_merge(self.node)
+ def analyzeTime(self, target_date: date) -> Dict[str, float]:
+ """:returns: A dictionary with keys as the asset class names and values as the
+ sum of investments corresponding to each class."""
+ return self._recursive_mergeTime(self.node, target_date)
+
+ def _recursive_mergeTime(self, node: Node, target_date: date) -> Dict[str, Any]:
+ """Internal method for recursive searching."""
+ total = {}
+
+ # Lines simply return their own amount
+ if isinstance(node, Line):
+ if node.envelope:
+ total[node.envelope.name] = node.get_amount()
+ else:
+ total["Unknown"] = node.get_amount()
+ return total
+
+ # Folders merge what the children return
+ elif isinstance(node, Folder):
+ for child in node.children:
+ for key, value in self._recursive_mergeTime(child,target_date).items():
+ if key in total.keys():
+ total[key] += value
+ else:
+ total[key] = value
+ return total
+
+ # Safeguard for future versions
+ else:
+ raise ValueError(f"Unknown node type '{type(node)}'.")
+
def _recursive_merge(self, node: Node) -> Dict[str, float]:
"""Internal method for recursive searching."""
total = {}
diff --git a/finalynx/analyzer/lines.py b/finalynx/analyzer/lines.py
new file mode 100644
index 0000000..7449f5a
--- /dev/null
+++ b/finalynx/analyzer/lines.py
@@ -0,0 +1,47 @@
+from datetime import date
+from typing import Any
+from typing import Dict
+
+from ..portfolio import Folder
+from ..portfolio import Line
+from ..portfolio import Node
+from .analyzer import Analyzer
+
+
+class AnalyzeLines(Analyzer):
+ """Aims to agglomerate the children's pf lines and return
+ the amount represented by each line.
+ :returns: a dictionary with lines as keys and the
+ corresponding total amount contained in the children.
+ """
+
+ def analyzeTime(self, target_date: date) -> Dict[str, float]:
+ """:returns: A dictionary with keys as the asset class names and values as the
+ sum of investments corresponding to each class."""
+ return self._recursive_mergeTime(self.node, target_date)
+
+ def _recursive_mergeTime(self, node: Node, target_date: date) -> Dict[str, Any]:
+ """Internal method for recursive searching."""
+ total = {}
+
+ # Lines simply return their own amount
+ if isinstance(node, Line):
+ if node.name:
+ total[node.name] = node.get_amount()
+ else:
+ total["Unknown"] = node.get_amount()
+ return total
+
+ # Folders merge what the children return
+ elif isinstance(node, Folder):
+ for child in node.children:
+ for key, value in self._recursive_mergeTime(child,target_date).items():
+ if key in total.keys():
+ total[key] += value
+ else:
+ total[key] = value
+ return total
+
+ # Safeguard for future versions
+ else:
+ raise ValueError(f"Unknown node type '{type(node)}'.")
diff --git a/finalynx/analyzer/subasset_class.py b/finalynx/analyzer/subasset_class.py
new file mode 100644
index 0000000..daf67bf
--- /dev/null
+++ b/finalynx/analyzer/subasset_class.py
@@ -0,0 +1,165 @@
+from datetime import date
+from typing import Any
+from typing import Dict
+
+import numpy as np
+
+from ..portfolio import AssetClass
+from ..portfolio import AssetSubclass
+from ..portfolio import Folder
+from ..portfolio import Line
+from ..portfolio import Node
+from .analyzer import Analyzer
+
+
+class AnalyzeSubAssetClasses(Analyzer):
+ """Aims to agglomerate the children's Sub asset classes and return
+ the amount represented by each Sub asset class.
+ :returns: a dictionary with Sub asset classes as keys and the
+ corresponding total amount contained in the children.
+ """
+
+ SUBASSET_COLORS_FINARY = {
+ # Cash
+ "Comptes courants": "#eed7b4",
+ "Monétaire": "#eed7b4",
+ "Liquidités": "#eed7b4",
+ # Guaranteed investments (mostly french)
+ "Livrets": "#b966f5",
+ "Livrets imposables": "#b966f5",
+ "Fonds euro": "#b966f5",
+ # Bonds
+ "Fonds datés": "#87bc45",
+ # Stocks
+ "Titres vifs": "#3a84de",
+ "ETF": "#3a84de",
+ # Real estate
+ "Immobilier physique": "#deab5e",
+ "SCPI": "#deab5e",
+ "SCI": "#deab5e",
+ # Metals
+ "Or": "#77cfac",
+ "Argent": "#77cfac",
+ "Matières premières": "#77cfac",
+ # Cryptos
+ "L1": "#bdcf32",
+ "Stablecoins": "#bdcf32",
+ "DeFi": "#bdcf32",
+ # Passives
+ "Véhicule": "#434348",
+ "Passif": "#434348",
+ # Exotics
+ "Forêts": "#228c83",
+ "Art": "#228c83",
+ "Watches": "#228c83",
+ "Crowdlending": "#228c83",
+ "Startup": "#228c83",
+ # Diversified
+ "Diversifié": "#b54093",
+ "OPCVM": "#b54093",
+ # Unknown (default)
+ "Unknown": "#b54053",
+ }
+
+ SUBASSET_COLORS_CUSTOM = {
+ # Cash
+ "Comptes courants": "#eed7b4",
+ "Monétaire": "#eed7b4",
+ "Liquidités": "#eed7b4",
+ # Guaranteed investments (mostly french)
+ "Livrets": "#b966f5",
+ "Livrets imposables": "#b966f5",
+ "Fonds euro": "#b966f5",
+ # Bonds
+ "Fonds datés": "#87bc45",
+ # Stocks
+ "Titres vifs": "#3a84de",
+ "ETF": "#3a84de",
+ # Real estate
+ "Immobilier physique": "#deab5e",
+ "SCPI": "#deab5e",
+ "SCI": "#deab5e",
+ # Metals
+ "Or": "#77cfac",
+ "Argent": "#77cfac",
+ "Matières premières": "#77cfac",
+ # Cryptos
+ "L1": "#bdcf32",
+ "Stablecoins": "#bdcf32",
+ "DeFi": "#bdcf32",
+ # Passives
+ "Véhicule": "#434348",
+ "Passif": "#434348",
+ # Exotics
+ "Forêts": "#228c83",
+ "Art": "#228c83",
+ "Watches": "#228c83",
+ "Crowdlending": "#228c83",
+ "Startup": "#228c83",
+ # Diversified
+ "Diversifié": "#b54093",
+ "OPCVM": "#b54093",
+ # Unknown (default)
+ "Unknown": "#b54053",
+ }
+
+ def analyze(self) -> Dict[str, Any]:
+ """:returns: A dictionary with keys as the asset class names and values as the
+ sum of investments corresponding to each class."""
+ return self._recursive_merge(self.node)
+
+ def analyzeTime(self, target_date: date) -> Dict[str, float]:
+ """:returns: A dictionary with keys as the Sub asset class names and values as the
+ sum of investments corresponding to each class."""
+ return self._recursive_mergeTime(self.node, target_date)
+
+ def _recursive_mergeTime(self, node: Node, target_date: date) -> Dict[str, Any]:
+ """Internal method for recursive searching."""
+ total = {}
+
+ # Lines simply return their own amount
+ if isinstance(node, Line):
+ total[node.asset_subclass.value] = node.get_amount()
+ return total
+
+ # Folders merge what the children return
+ elif isinstance(node, Folder):
+ for child in node.children:
+ for key, value in self._recursive_mergeTime(child, target_date).items():
+ if key in total.keys():
+ total[key] += value
+ else:
+ total[key] = value
+ # for subkey, subvalue in value["subclasses"].items():
+ # total[subkey] += subvalue
+ return total
+
+ # Safeguard for future versions
+ else:
+ raise ValueError(f"Unknown node type '{type(node)}'.")
+
+ def _recursive_merge(self, node: Node) -> Dict[str, Any]:
+ """Internal method for recursive searching."""
+ result: Dict[str, Any] = {
+ c.value: {"total": 0.0, "subclasses": {s.value: 0.0 for s in AssetSubclass}} for c in AssetClass
+ }
+
+ # Lines simply return their own amount
+ if isinstance(node, Line):
+ result[node.asset_class.value]["total"] = node.get_amount()
+ result[node.asset_class.value]["subclasses"][node.asset_subclass.value] = node.get_amount()
+ return result
+
+ # Folders merge what the children return
+ elif isinstance(node, Folder):
+ for child in node.children:
+ for key, subdict in self._recursive_merge(child).items():
+ result[key]["total"] += subdict["total"]
+
+ for subkey, subvalue in subdict["subclasses"].items():
+ result[key]["subclasses"][subkey] += subvalue
+ return result
+
+ # Safeguard for future versions
+ else:
+ raise ValueError(f"Unknown node type '{type(node)}'.")
diff --git a/finalynx/assistant.py b/finalynx/assistant.py
index ffd62a1..ff14eb5 100644
--- a/finalynx/assistant.py
+++ b/finalynx/assistant.py
@@ -197,6 +197,8 @@ def _parse_args(self) -> None:
self.simulation.print_each_step = True
if args["--sim-steps"] and self.simulation:
self.simulation.step_years = int(args["--sim-steps"])
+ if args["--metric-frequency"] and self.simulation:
+ self.simulation.metrics_record_freqency = str(args["--metric-frequency"])
if args["--theme"]:
theme_name = str(args["--theme"])
if theme_name not in finalynx.theme.AVAILABLE_THEMES:
diff --git a/finalynx/dashboard/dashboard.py b/finalynx/dashboard/dashboard.py
index 31e47fb..1968a87 100644
--- a/finalynx/dashboard/dashboard.py
+++ b/finalynx/dashboard/dashboard.py
@@ -12,6 +12,7 @@
from finalynx.analyzer.asset_class import AnalyzeAssetClasses
from finalynx.analyzer.envelopes import AnalyzeEnvelopes
from finalynx.analyzer.investment_state import AnalyzeInvestmentStates
+from finalynx.analyzer.subasset_class import AnalyzeSubAssetClasses
from finalynx.portfolio.folder import Folder
from finalynx.portfolio.folder import FolderDisplay
from finalynx.portfolio.line import Line
@@ -181,7 +182,23 @@ def _on_select_color_map(data: Any) -> None:
)
with ui.row():
self.chart_envelopes = ui.chart(AnalyzeEnvelopes(self.selected_node).chart())
- self.chart_simulation = ui.chart(timeline.chart() if timeline else {})
+ #self.chart_simulation = ui.chart(timeline.chart() if timeline else {})
+ self.chart_etats_enveloppes = ui.chart(timeline.chartOnTimeline("Evolution des états d'enveloppes",timeline._log_env_states,{
+ "Unknown": "#434348",
+ "Closed": "#999999",
+ "Locked": "#F94144",
+ "Taxed": "#F9C74F",
+ "Free": "#7BB151",
+ }) if timeline else {})
+ #self.chart_simulation1 = ui.chart(timeline.chartEnveloppeTimeline() if timeline else {})
+ self.chart_enveloppes = ui.chart(timeline.chartOnTimeline("Evolution des enveloppes",timeline._log_enveloppe_values) if timeline else {})
+ #self.chart_simulation2 = ui.chart(timeline.OLDchartBucket() if timeline else {})
+ #self.chart_simulation3 = ui.chart(timeline.chartAssetTimeline() if timeline else {})
+ self.chart_asset_classes = ui.chart(timeline.chartOnTimeline("Evolution des classes d'actifs",timeline._log_assets_classes_values,AnalyzeAssetClasses.ASSET_COLORS_FINARY) if timeline else {})
+ self.chart_subasset_classes = ui.chart(timeline.chartOnTimeline("Evolution des sous-classes d'actifs",timeline._log_assets_subclasses_values,AnalyzeSubAssetClasses.SUBASSET_COLORS_FINARY) if timeline else {})
+ #self.chart_simulation4 = ui.chart(timeline.chartSubAssetTimeline() if timeline else {})
+ self.chart_lines = ui.chart(timeline.chartOnTimeline("Evolution des lignes du portefeuille", timeline._log_lines_values, visible_by_default=False) if timeline else {})
+ #self.chart_simulation5 = ui.chart(timeline.chartLinesTimeline() if timeline else {})
ui.run(
title="Finalynx Dashboard",
diff --git a/finalynx/portfolio/constants.py b/finalynx/portfolio/constants.py
index aecc52b..f767794 100644
--- a/finalynx/portfolio/constants.py
+++ b/finalynx/portfolio/constants.py
@@ -93,6 +93,7 @@ class AssetSubclass(Enum):
# Passives
VEHICLE = "Véhicule"
+ PASSIVE = "Passif"
# Unknown (default)
UNKNOWN = "Unknown"
diff --git a/finalynx/simulator/timeline.py b/finalynx/simulator/timeline.py
index c452da2..4a43830 100644
--- a/finalynx/simulator/timeline.py
+++ b/finalynx/simulator/timeline.py
@@ -1,13 +1,18 @@
from dataclasses import dataclass
-from datetime import date
+from datetime import date, datetime
from datetime import timedelta
from typing import Any
from typing import Dict
from typing import List
from typing import Optional
+from finalynx.analyzer.asset_class import AnalyzeAssetClasses
+from finalynx.analyzer.envelopes import AnalyzeEnvelopes
+from finalynx.analyzer.lines import AnalyzeLines
from finalynx.analyzer.investment_state import AnalyzeInvestmentStates
+from finalynx.analyzer.subasset_class import AnalyzeSubAssetClasses
from finalynx.portfolio.bucket import Bucket
+from finalynx.portfolio.constants import AssetClass
from finalynx.portfolio.envelope import EnvelopeState
from finalynx.portfolio.folder import Portfolio
from finalynx.simulator.actions import AutoBalance
@@ -35,12 +40,14 @@ class Simulation:
# Whether to print the final portfolio state in the console after the simulation
print_final: bool = False
- # Whether to print the final portfolio state in the console after the simulation
+ # Whether to print the portfolio state in the console on each step of the simulation
print_each_step: bool = False
# Display the portfolio's worth in the console every `step` years
step_years: int = 5
+ # Record the portfolio stats on each day of the simulation 'DAY', 'MONTH', 'YEAR'
+ metrics_record_freqency: str = "MONTH"
class Timeline:
"""Main simulation engine to execute programmed actions on your portfolio."""
@@ -74,6 +81,11 @@ def __init__(
# Log some metrics during the simulation to display them at the end
self._log_dates: List[date] = [] # Dates at which the portfolio metrics were logged
self._log_env_states: Dict[str, List[float]] = {c.value: [] for c in EnvelopeState}
+ self._log_enveloppe_values: Dict[str, List[float]] = {}
+ self._log_assets_classes_values: Dict[str, List[float]] = {c.value: [] for c in AssetClass}
+ self._log_assets_subclasses_values: Dict[str, List[float]] = {}
+ self._log_lines_values: Dict[str, List[float]] = {}
+ self._log_events: Dict[date, List[str]] = {}
def run(self) -> None:
"""Step all events until the simulation limit is reached."""
@@ -93,6 +105,9 @@ def step_until(self, target_date: date) -> None:
"""Execute all events until the specified date is reached."""
assert self.current_date < target_date, "Target date must be in the future."
+ #Enregistrement de la situation de démarrage du Portefeuille
+ self._record_metrics()
+
while self.current_date < target_date and not self.is_finished:
if self.step():
return
@@ -111,6 +126,10 @@ def step(self) -> bool:
# Add the newly generated events and sort the event list by date
new_events = next_event.apply(self._portfolio)
+ if next_event.planned_date in self._log_events:
+ self._log_events[next_event.planned_date].append(next_event.name)
+ else:
+ self._log_events[next_event.planned_date]=[next_event.name]
# Recalculate the amounts for shared folders
for bucket in self._buckets:
@@ -123,7 +142,10 @@ def step(self) -> bool:
self._sort_events()
# Record the metrics if the year changed
- if next_event.planned_date.year != self.current_date.year:
+ #if next_event.planned_date.year != self.current_date.year:
+ # self._record_metrics()
+ if (self.simulation.metrics_record_freqency == "DAY" and next_event.planned_date != self.current_date) or (self.simulation.metrics_record_freqency == "YEAR" and next_event.planned_date.year != self.current_date.year) or (self.simulation.metrics_record_freqency == "MONTH" and next_event.planned_date.month != self.current_date.month):
+ self.current_date = next_event.planned_date
self._record_metrics()
# Move the current date to this event's date
@@ -146,29 +168,65 @@ def is_finished(self) -> bool:
def _record_metrics(self) -> None:
"""Record the portfolio's metrics at the current date to display later."""
- self._log_dates.append(self.current_date)
-
- # Record the envelope states and their amounts at this date
- for key, value in AnalyzeInvestmentStates(self._portfolio).analyze(self.current_date).items():
- self._log_env_states[key].append(value)
-
- def chart(self) -> Dict[str, Any]:
- """Plot a Highcharts chart of the portfolio's envelopes' states and amounts over time."""
- assert self._log_env_states, "Run the simulation before charting."
-
- colors = {
- "Unknown": "#434348",
- "Closed": "#999999",
- "Locked": "#F94144",
- "Taxed": "#F9C74F",
- "Free": "#7BB151",
- }
+ if self.current_date not in self._log_dates:
+ self._log_dates.append(self.current_date)
+
+ # Record the envelope states and their amounts at this date
+ for key, value in AnalyzeInvestmentStates(self._portfolio).analyze(self.current_date).items():
+ self._log_env_states[key].append(value)
+
+ for key, value in AnalyzeEnvelopes(self._portfolio).analyzeTime(self.current_date).items():
+ if key in self._log_enveloppe_values:
+ self._log_enveloppe_values[key].append(value)
+ else:
+ self._log_enveloppe_values[key]=[value]
+
+ for key, value in AnalyzeAssetClasses(self._portfolio).analyzeTime(self.current_date).items():
+ self._log_assets_classes_values[key].append(value)
+
+ for key, value in AnalyzeSubAssetClasses(self._portfolio).analyzeTime(self.current_date).items():
+ if key in self._log_assets_subclasses_values:
+ self._log_assets_subclasses_values[key].append(value)
+ else:
+ self._log_assets_subclasses_values[key]=[value]
+
+ for key, value in AnalyzeLines(self._portfolio).analyzeTime(self.current_date).items():
+ if key in self._log_lines_values:
+ self._log_lines_values[key].append(value)
+ else:
+ self._log_lines_values[key]=[value]
+ else:
+ #On doit remplacer les valeurs stockées par les nouvelles sans créer 2 fois l'enregistrement
+ ident = self._log_dates.index(self.current_date)
+ # Record the envelope states and their amounts at this date
+ for key, value in AnalyzeInvestmentStates(self._portfolio).analyze(self.current_date).items():
+ self._log_env_states[key][-1]=value
+ for key, value in AnalyzeEnvelopes(self._portfolio).analyzeTime(self.current_date).items():
+ if key in self._log_enveloppe_values:
+ self._log_enveloppe_values[key][-1]=value
+ else:
+ self._log_enveloppe_values[key]=[value]
+ for key, value in AnalyzeAssetClasses(self._portfolio).analyzeTime(self.current_date).items():
+ self._log_assets_classes_values[key][-1]=value
+ for key, value in AnalyzeSubAssetClasses(self._portfolio).analyzeTime(self.current_date).items():
+ if key in self._log_assets_subclasses_values:
+ self._log_assets_subclasses_values[key][-1]=value
+ else:
+ self._log_assets_subclasses_values[key]=[value]
+ for key, value in AnalyzeLines(self._portfolio).analyzeTime(self.current_date).items():
+ if key in self._log_lines_values:
+ self._log_lines_values[key][-1]=value
+ else:
+ self._log_lines_values[key]=[value]
+
+ def chartOnTimeline(self, title: str, valuesToGraph: Dict[str, List[float]], colors: Dict[str,str]={}, visible_by_default: bool = True) -> Dict[str, Any]:
+ """Plot a Highcharts chart of the portfolio's caracteristics and amounts over time."""
+ #assert self._log_enveloppe_values, "Run the simulation before charting."
return {
- "chart": {"plotBackgroundColor": None, "plotBorderWidth": None, "plotShadow": False, "type": "area"},
- "title": {"text": "Simulation", "align": "center"},
+ "chart": {"plotBackgroundColor": None, "plotBorderWidth": None, "plotShadow": False, "type": "area", "zooming": {"type": 'xy' }, "height":1200, "width":1000},
+ "title": {"text": title, "align": "center"},
"plotOptions": {
- "series": {"pointStart": self._log_dates[0].year},
"area": {
"stacking": "normal",
"lineColor": "#666666",
@@ -179,14 +237,39 @@ def chart(self) -> Dict[str, Any]:
"series": [
{
"name": key,
- "data": value,
- "color": colors[key],
+ "data": self.convertDataSeries(value),
+ "visible": visible_by_default,
+ "color": colors[key] if (key in colors) else {None}
}
- for key, value in self._log_env_states.items()
+ for key, value in valuesToGraph.items()
],
+ "xAxis": {"type": 'datetime'},
+ "yAxis": {"crosshair": True},
+ "tooltip": {
+ "xDateFormat": '%m %Y',
+ "pointFormat": '{point.x:%e/%m/%Y}: {point.y:,.0f}€
',
+ "footerFormat": '{series.name}',
+ },
"credits": {"enabled": False},
}
+ def convertDataSeries(self, data: [float])->[Any]:
+ """Convert DataSeries in a time series format to allow non regular data"""
+ res=[]
+ i = 0
+ while i < len(data):
+ if (self._log_dates[i] in self._log_events):
+ evenements = "* "+"
* ".join(self._log_events[self._log_dates[i]])
+ else:
+ evenements = ""
+ point = {"x": datetime.combine(self._log_dates[i], datetime.min.time()).timestamp() * 1000,
+ "y": data[i],
+ "name": evenements
+ }
+ res.append(point)
+ i+=1
+ return res
+
def _sort_events(self) -> None:
"""Internal method to sort the event list by planned date."""
self._events.sort(key=lambda event: event.planned_date)
diff --git a/finalynx/usage.py b/finalynx/usage.py
index fb393be..1259a59 100644
--- a/finalynx/usage.py
+++ b/finalynx/usage.py
@@ -57,5 +57,6 @@ def main_filter(message: str) -> str:
--sim-steps=int Display the simulated portfolio's worth every X years, defaults to 5
--future Print the portfolio after the simulation has finished
--each-step Print the portfolio for each step of the simulation
+ --metric-frequency Record the portfolio stats on each day of the simulation 'DAY', 'MONTH', 'YEAR'
"""