portfolio-engine / chart_data.py
engineportf's picture
Initial Deployment from Local Engine
208fbf8 verified
Raw
History Blame Contribute Delete
1.78 kB
import pandas as pd
import numpy as np
import json
import datetime
import decimal
def _ds(series, n=250):
"""Downsamples a time series to a maximum of `n` points for frontend rendering speed."""
if series is None:
return series
# Note: Ensure it's a pandas Series to prevent .iloc crashes on raw lists/arrays
if not isinstance(series, (pd.Series, pd.DataFrame)):
series = pd.Series(series)
elif not series.index.is_unique:
# Note: Deduplicate index to prevent Chart.js from rendering zig-zag lines
series = series[~series.index.duplicated(keep='last')]
if len(series) <= n:
return series
# Note: Use uniform np.linspace to avoid discontinuity seams from concatenating tails
idx = np.round(np.linspace(0, len(series) - 1, n)).astype(int)
return series.iloc[idx]
def safe_json(obj):
"""
Safely serializes deep dictionaries containing NumPy/Pandas native types,
Datetimes, NaNs, and NaTs for Chart.js parsing.
"""
def _fix(o):
if o is None or o is pd.NA or o is pd.NaT:
return None
if isinstance(o, decimal.Decimal):
return float(o)
if isinstance(o, (float, np.floating)):
return None if not np.isfinite(o) else float(o)
if isinstance(o, (int, np.integer)):
return int(o)
if isinstance(o, (pd.Timestamp, datetime.datetime, datetime.date)):
return str(o)
if isinstance(o, dict):
return {k: _fix(v) for k, v in o.items()}
if isinstance(o, (list, tuple, np.ndarray, pd.Series, pd.Index)):
return [_fix(v) for v in o]
return o
return json.dumps(_fix(obj))