crypto_trading_agent / memory_utils.py
samsonleegh's picture
Upload 7 files
f5ed163 verified
# memory_utils.py
import json
import pandas as pd
from datetime import datetime
from pathlib import Path
from typing import Optional, List, Dict, Any
MEMORY_FILE = Path("memories.json")
def load_memories(n: int = 3) -> list[str]:
"""
Load the last n summarized memories from the JSONL file.
Returns a list of summary strings.
"""
if not MEMORY_FILE.exists():
return []
with open(MEMORY_FILE, "r", encoding="utf-8") as f:
lines = [json.loads(l) for l in f if l.strip()]
return [m["summary"] for m in lines[-n:]]
def save_memory(summary: str):
"""
Append a new memory summary to the JSONL file.
Each line is a JSON object: {"timestamp": "...", "summary": "..."}
"""
entry = {"timestamp": datetime.now().isoformat(), "summary": summary}
with open(MEMORY_FILE, "a", encoding="utf-8") as f:
f.write(json.dumps(entry) + "\n")
def show_memories(n: int = 5) -> str:
"""
Return a formatted string showing the latest n memories.
Useful for displaying in Gradio or CLI.
"""
memories = load_memories(n)
if not memories:
return "No memories yet."
formatted = "\n\n".join(
[f"๐Ÿ•’ Memory {i+1}: {m}" for i, m in enumerate(memories[::-1])]
)
return formatted
def load_memories_df(n: Optional[int] = None):
"""
Return recent memories as a pandas DataFrame (newest first).
"""
with open("memories.json", "r", encoding="utf-8") as f:
lines = [json.loads(line) for line in f if line.strip()]
df = pd.DataFrame(lines)
return df.tail(n).iloc[::-1] if n else df.iloc[::-1]