diff --git a/cli/main.py b/cli/main.py index 2e06d50c..83c0f520 100644 --- a/cli/main.py +++ b/cli/main.py @@ -1,5 +1,6 @@ from typing import Optional import datetime +import os import typer from pathlib import Path from functools import wraps @@ -398,7 +399,7 @@ def update_display(layout, spinner_text=None): def get_user_selections(): """Get all user selections before starting the analysis display.""" # Display ASCII art welcome message - with open("./cli/static/welcome.txt", "r") as f: + with open("./cli/static/welcome.txt", "r", encoding="utf-8") as f: welcome_ascii = f.read() # Create welcome box content @@ -748,11 +749,6 @@ def run_analysis(): config["backend_url"] = selections["backend_url"] config["llm_provider"] = selections["llm_provider"].lower() - # Initialize the graph - graph = TradingAgentsGraph( - [analyst.value for analyst in selections["analysts"]], config=config, debug=True - ) - # Create result directory results_dir = Path(config["results_dir"]) / selections["ticker"] / selections["analysis_date"] results_dir.mkdir(parents=True, exist_ok=True) @@ -761,6 +757,18 @@ def run_analysis(): log_file = results_dir / "message_tool.log" log_file.touch(exist_ok=True) + # ACE skillbook + ace_skillbook_path = os.path.join(config.get("results_dir", "./results"), "ace_skillbook.json") + + # Initialize the graph with ACE enabled + graph = TradingAgentsGraph( + [analyst.value for analyst in selections["analysts"]], + config=config, + debug=True, + ace_enabled=True, + ace_skillbook_path=ace_skillbook_path, + ) + def save_message_decorator(obj, func_name): func = getattr(obj, func_name) @wraps(func) @@ -768,7 +776,7 @@ def run_analysis(): func(*args, **kwargs) timestamp, message_type, content = obj.messages[-1] content = content.replace("\n", " ") # Replace newlines with spaces - with open(log_file, "a") as f: + with open(log_file, "a", encoding="utf-8") as f: f.write(f"{timestamp} [{message_type}] {content}\n") return wrapper @@ -779,7 +787,7 @@ def run_analysis(): func(*args, **kwargs) timestamp, tool_name, args = obj.tool_calls[-1] args_str = ", ".join(f"{k}={v}" for k, v in args.items()) - with open(log_file, "a") as f: + with open(log_file, "a", encoding="utf-8") as f: f.write(f"{timestamp} [Tool Call] {tool_name}({args_str})\n") return wrapper @@ -792,7 +800,7 @@ def run_analysis(): content = obj.report_sections[section_name] if content: file_name = f"{section_name}.md" - with open(report_dir / file_name, "w") as f: + with open(report_dir / file_name, "w", encoding="utf-8") as f: f.write(content) return wrapper @@ -1079,6 +1087,17 @@ def run_analysis(): # Get final state and decision final_state = trace[-1] + graph.curr_state = final_state # Ensure curr_state is set for ACE + + # Trigger ACE learning from analysis (price-based) + if graph.ace_enabled: + message_buffer.add_message("ACE", "Learning from analysis...") + try: + graph._ace_learn_from_analysis() + message_buffer.add_message("ACE", "Learning cycle completed.") + except Exception as e: + message_buffer.add_message("ACE", f"Learning failed: {e}") + decision = graph.process_signal(final_state["final_trade_decision"]) # Update all agent statuses to completed @@ -1089,6 +1108,14 @@ def run_analysis(): "Analysis", f"Completed analysis for {selections['analysis_date']}" ) + # Save ACE skillbook + if graph.ace_engine: + try: + saved_path = graph.save_ace_skillbook() + message_buffer.add_message("ACE", f"Skillbook saved to {saved_path}") + except Exception as e: + message_buffer.add_message("ACE", f"Failed to save skillbook: {e}") + # Update final report sections for section in message_buffer.report_sections.keys(): if section in final_state: diff --git a/requirements.txt b/requirements.txt index a6154cd2..25216d29 100644 --- a/requirements.txt +++ b/requirements.txt @@ -24,3 +24,5 @@ rich questionary langchain_anthropic langchain-google-genai +python-dotenv +ace-framework diff --git a/tradingagents/ace/__init__.py b/tradingagents/ace/__init__.py new file mode 100644 index 00000000..0db938d3 --- /dev/null +++ b/tradingagents/ace/__init__.py @@ -0,0 +1,36 @@ +""" +Agentic Context Engineering (ACE) implementation for TradingAgents. + +Uses the official Kayba ACE framework (pip install ace-framework). +Based on the ACE paper (arXiv:2510.04618) - enables agents to improve through +in-context learning instead of fine-tuning. + +Core pattern: +1. INJECT: Add learned strategies to agent prompts +2. EXECUTE: Agent performs task using accumulated knowledge +3. LEARN: Reflector analyzes results, SkillManager updates skillbook +""" + +from ace import ( + ACELiteLLM, + Skillbook, + Skill, + Reflector, + SkillManager, + UpdateOperation, + UpdateBatch, +) + +from .kayba_ace import TradingACE, create_trading_ace + +__all__ = [ + "ACELiteLLM", + "Skillbook", + "Skill", + "Reflector", + "SkillManager", + "UpdateOperation", + "UpdateBatch", + "TradingACE", + "create_trading_ace", +] \ No newline at end of file diff --git a/tradingagents/ace/kayba_ace.py b/tradingagents/ace/kayba_ace.py new file mode 100644 index 00000000..644421a3 --- /dev/null +++ b/tradingagents/ace/kayba_ace.py @@ -0,0 +1,223 @@ +""" +Kayba ACE Integration for TradingAgents. + +Uses the official ace-framework from Kayba (pip install ace-framework) +for self-improving trading agents. +""" + +from __future__ import annotations + +import os +from pathlib import Path +from typing import Any, Dict, List, Optional + +from ace import ( + OnlineACE, + Agent, + Reflector, + SkillManager, + LiteLLMClient, + Sample, + TaskEnvironment, + EnvironmentResult, + Skillbook, + Skill, + AgentOutput +) + + +class TradingEnvironment(TaskEnvironment): + """ + Environment for evaluating the quality and consistency of trading analysis. + + Instead of just looking at price, it evaluates the logical flow between + market data, sentiment, news, and the final decision. + """ + + def evaluate(self, sample: Sample, agent_output) -> EnvironmentResult: + """ + Evaluate the analytical rigor of the trading decision. + """ + # We provide a high-level goal for the reflector + feedback = ( + "Evaluate the logical consistency of this analysis. " + "Check if the final decision is truly supported by the market report, " + "sentiment analysis, and news. Identify any contradictions or missed " + "signals that could lead to a sub-optimal trade, regardless of the price outcome." + ) + + return EnvironmentResult( + feedback=feedback, + ground_truth="A logically sound, consistent, and well-supported investment thesis." + ) + + +class TradingACE: + """ + Self-improving trading agent using Kayba's ACE framework. + """ + + def __init__( + self, + model: str = "gpt-4o-mini", + skillbook_path: Optional[str] = None, + api_base: Optional[str] = None, + ): + """ + Initialize TradingACE. + """ + self.model = model + self.skillbook_path = skillbook_path + + # Initialize LLM client + self.client = LiteLLMClient(model=model, api_base=api_base) + + # Create ACE components + self.agent = Agent(self.client) + self.reflector = Reflector(self.client) + self.skill_manager = SkillManager(self.client) + + # Load or create skillbook + if skillbook_path and Path(skillbook_path).exists(): + self.skillbook = Skillbook.load_from_file(skillbook_path) + else: + self.skillbook = Skillbook() + + # Create OnlineACE adapter + self.ace = OnlineACE( + skillbook=self.skillbook, + agent=self.agent, + reflector=self.reflector, + skill_manager=self.skill_manager + ) + + self.environment = TradingEnvironment() + + def learn_from_analysis(self, reports: Dict[str, str], decision: str): + """ + Learn from a trading analysis by reflecting on the consistency of all reports. + """ + ticker = reports.get("ticker", "Unknown") + date = reports.get("date", "Unknown") + + print(f"ACE: Learning from analytical consistency for {ticker} on {date}") + + # Combine all reports into a single context for the reflector + full_context = "\n\n".join([ + f"MARKET REPORT:\n{reports.get('market', 'N/A')}", + f"SENTIMENT REPORT:\n{reports.get('sentiment', 'N/A')}", + f"NEWS REPORT:\n{reports.get('news', 'N/A')}", + f"FUNDAMENTALS REPORT:\n{reports.get('fundamentals', 'N/A')}", + f"INVESTMENT PLAN:\n{reports.get('plan', 'N/A')}" + ]) + + sample = Sample( + question=( + f"Analyze the following multi-agent trading reports for {ticker} and " + "determine if the final decision is logically consistent with all data points." + ), + context=full_context, + ground_truth="A perfectly consistent and well-reasoned investment thesis." + ) + + # Create a proper AgentOutput instance representing the whole analysis + actual_output = AgentOutput( + reasoning=f"Full analysis process for {ticker}.", + final_answer=decision, + raw={"decision": decision} + ) + + try: + # 1. Evaluate the analytical rigor + eval_result = self.environment.evaluate(sample, actual_output) + print(f"ACE: Evaluation focus: {eval_result.feedback}") + + # 2. Reflect on the consistency and quality + print("ACE: Reflecting on analytical quality...") + reflector_output = self.reflector.reflect( + question=sample.question, + agent_output=actual_output, + skillbook=self.skillbook, + ground_truth=eval_result.ground_truth, + feedback=eval_result.feedback + ) + + print(f"ACE: Reflection generated (reasoning length: {len(reflector_output.reasoning)})") + + # 3. Update the skillbook with the new reflection + print("ACE: Updating skillbook...") + sm_output = self.skill_manager.update_skills( + skillbook=self.skillbook, + reflection=reflector_output, + question_context=sample.context, + progress=eval_result.feedback + ) + + if sm_output.update: + print("ACE: Applying update to skillbook...") + self.skillbook.apply_update(sm_output.update) + + # Force save + self.save_skillbook() + print(f"ACE: Skillbook saved to {self.skillbook_path}") + + except Exception as e: + print(f"Error in ACE learning: {e}") + import traceback + traceback.print_exc() + + def learn_from_trade(self, context: str, decision: str, result: str, market_data: str): + """ + Compatibility method for TradingAgentsGraph. + """ + reports = { + "market": market_data, + "ticker": context.split(" on ")[0], + "date": context.split(" on ")[1] if " on " in context else "Unknown" + } + self.learn_from_analysis(reports, decision) + + def get_skills_context(self) -> str: + """Get formatted skills for injection into prompts.""" + if not self.skillbook.skills(): + return "" + return self.skillbook.as_prompt() + + def save_skillbook(self, path: Optional[str] = None) -> str: + """Save the skillbook to file.""" + save_path = path or self.skillbook_path or "ace_skillbook.json" + self.skillbook.save_to_file(save_path) + return save_path + + def get_stats(self) -> Dict[str, Any]: + """Get ACE statistics.""" + try: + skills = self.skillbook.skills() + count = len(skills) + except: + try: + count = self.skillbook.stats().get('skills', 0) + except: + count = 0 + + return { + "skills_count": count, + "model": self.model + } + + +def create_trading_ace( + config: Dict[str, Any], + skillbook_path: Optional[str] = None, +) -> TradingACE: + """ + Factory function to create TradingACE. + """ + model = config.get("quick_think_llm", "gpt-4o-mini") + api_base = config.get("backend_url") + + return TradingACE( + model=model, + skillbook_path=skillbook_path, + api_base=api_base + ) \ No newline at end of file diff --git a/tradingagents/agents/trader/trader.py b/tradingagents/agents/trader/trader.py index 1b05c35d..46800294 100644 --- a/tradingagents/agents/trader/trader.py +++ b/tradingagents/agents/trader/trader.py @@ -3,7 +3,7 @@ import time import json -def create_trader(llm, memory): +def create_trader(llm, memory, ace_context=""): def trader_node(state, name): company_name = state["company_of_interest"] investment_plan = state["investment_plan"] @@ -22,9 +22,14 @@ def create_trader(llm, memory): else: past_memory_str = "No past memories found." + # Add ACE context if available + ace_str = "" + if ace_context: + ace_str = f"\n\nLearned Trading Strategies (ACE):\n{ace_context}" + context = { "role": "user", - "content": f"Based on a comprehensive analysis by a team of analysts, here is an investment plan tailored for {company_name}. This plan incorporates insights from current technical market trends, macroeconomic indicators, and social media sentiment. Use this plan as a foundation for evaluating your next trading decision.\n\nProposed Investment Plan: {investment_plan}\n\nLeverage these insights to make an informed and strategic decision.", + "content": f"Based on a comprehensive analysis by a team of analysts, here is an investment plan tailored for {company_name}. This plan incorporates insights from current technical market trends, macroeconomic indicators, and social media sentiment. Use this plan as a foundation for evaluating your next trading decision.\n\nProposed Investment Plan: {investment_plan}\n\nLeverage these insights to make an informed and strategic decision.{ace_str}", } messages = [ diff --git a/tradingagents/default_config.py b/tradingagents/default_config.py index 1f40a2a2..1da519ea 100644 --- a/tradingagents/default_config.py +++ b/tradingagents/default_config.py @@ -17,6 +17,9 @@ DEFAULT_CONFIG = { "max_debate_rounds": 1, "max_risk_discuss_rounds": 1, "max_recur_limit": 100, + # ACE (Agentic Context Engineering) settings + "ace_enabled": os.getenv("ACE_ENABLED", "true").lower() in {"true", "1", "yes", "on"}, + "ace_skillbook_path": os.getenv("ACE_SKILLBOOK_PATH", "results/ace_skillbook.json"), # Data vendor configuration # Category-level configuration (default for all tools in category) "data_vendors": { diff --git a/tradingagents/graph/setup.py b/tradingagents/graph/setup.py index b270ffc0..65ff2f94 100644 --- a/tradingagents/graph/setup.py +++ b/tradingagents/graph/setup.py @@ -25,6 +25,7 @@ class GraphSetup: invest_judge_memory, risk_manager_memory, conditional_logic: ConditionalLogic, + ace_context: str = "", ): """Initialize with required components.""" self.quick_thinking_llm = quick_thinking_llm @@ -36,6 +37,7 @@ class GraphSetup: self.invest_judge_memory = invest_judge_memory self.risk_manager_memory = risk_manager_memory self.conditional_logic = conditional_logic + self.ace_context = ace_context def setup_graph( self, selected_analysts=["market", "social", "news", "fundamentals"] @@ -95,7 +97,7 @@ class GraphSetup: research_manager_node = create_research_manager( self.deep_thinking_llm, self.invest_judge_memory ) - trader_node = create_trader(self.quick_thinking_llm, self.trader_memory) + trader_node = create_trader(self.quick_thinking_llm, self.trader_memory, self.ace_context) # Create risk analysis nodes risky_analyst = create_risky_debator(self.quick_thinking_llm) diff --git a/tradingagents/graph/trading_graph.py b/tradingagents/graph/trading_graph.py index 40cdff75..8f4232ee 100644 --- a/tradingagents/graph/trading_graph.py +++ b/tradingagents/graph/trading_graph.py @@ -1,6 +1,7 @@ # TradingAgents/graph/trading_graph.py import os +import re from pathlib import Path import json from datetime import date @@ -21,6 +22,7 @@ from tradingagents.agents.utils.agent_states import ( RiskDebateState, ) from tradingagents.dataflows.config import set_config +from tradingagents.ace import TradingACE, create_trading_ace # Import the new abstract tool methods from agent_utils from tradingagents.agents.utils.agent_utils import ( @@ -42,6 +44,8 @@ from .propagation import Propagator from .reflection import Reflector from .signal_processing import SignalProcessor +from tradingagents.ace import TradingACE, create_trading_ace + class TradingAgentsGraph: """Main class that orchestrates the trading agents framework.""" @@ -51,6 +55,8 @@ class TradingAgentsGraph: selected_analysts=["market", "social", "news", "fundamentals"], debug=False, config: Dict[str, Any] = None, + ace_enabled: bool = True, + ace_skillbook_path: Optional[str] = None, ): """Initialize the trading agents graph and components. @@ -58,6 +64,8 @@ class TradingAgentsGraph: selected_analysts: List of analyst types to include debug: Whether to run in debug mode config: Configuration dictionary. If None, uses default config + ace_enabled: Whether to enable ACE learning (default: True) + ace_skillbook_path: Path to load/save ACE skillbook (optional) """ self.debug = debug self.config = config or DEFAULT_CONFIG @@ -91,6 +99,29 @@ class TradingAgentsGraph: self.invest_judge_memory = FinancialSituationMemory("invest_judge_memory", self.config) self.risk_manager_memory = FinancialSituationMemory("risk_manager_memory", self.config) + # Initialize ACE Engine + self.ace_enabled = ace_enabled + self.ace_skillbook_path = ace_skillbook_path or self.config.get( + "ace_skillbook_path", + os.path.join(self.config.get("results_dir", "./results"), "ace_skillbook.json") + ) + + if self.ace_enabled: + self.ace_engine = create_trading_ace( + config=self.config, + skillbook_path=self.ace_skillbook_path, + ) + else: + self.ace_engine = None + + if self.ace_enabled: + self.ace_engine = create_trading_ace( + config=self.config, + skillbook_path=self.ace_skillbook_path, + ) + else: + self.ace_engine = None + # Create tool nodes self.tool_nodes = self._create_tool_nodes() @@ -106,6 +137,7 @@ class TradingAgentsGraph: self.invest_judge_memory, self.risk_manager_memory, self.conditional_logic, + ace_context=self.get_ace_context(), ) self.propagator = Propagator() @@ -231,6 +263,7 @@ class TradingAgentsGraph: with open( f"eval_results/{self.ticker}/TradingAgentsStrategy_logs/full_states_log_{trade_date}.json", "w", + encoding="utf-8" ) as f: json.dump(self.log_states_dict, f, indent=4) @@ -252,6 +285,125 @@ class TradingAgentsGraph: self.curr_state, returns_losses, self.risk_manager_memory ) + # ACE Learning: Learn from trading execution + if self.ace_enabled and self.ace_engine and self.curr_state: + self._ace_learn(returns_losses) + + def _ace_learn_from_analysis(self): + """ + Trigger ACE learning based on the analytical consistency of all reports. + """ + if not self.ace_enabled or not self.ace_engine or not self.curr_state: + return + + print(f"DEBUG: ACE analytical reflection triggered for {self.curr_state.get('company_of_interest')}") + + reports = { + "ticker": self.curr_state.get("company_of_interest", "Unknown"), + "date": str(self.curr_state.get("trade_date", "Unknown")), + "market": self.curr_state.get("market_report", ""), + "sentiment": self.curr_state.get("sentiment_report", ""), + "news": self.curr_state.get("news_report", ""), + "fundamentals": self.curr_state.get("fundamentals_report", ""), + "plan": self.curr_state.get("investment_plan", ""), + } + + decision = self.curr_state.get("final_trade_decision", "") + + decision = re.sub(r"\[ACE_METADATA: .*\]", "", decision).strip() + + self.ace_engine.learn_from_analysis( + reports=reports, + decision=decision + ) + + def _ace_learn(self, returns_losses): + """Apply ACE learning from the current trading state using Kayba ACE.""" + if not self.curr_state: + return + + context = f"{self.curr_state['company_of_interest']} on {self.curr_state['trade_date']}" + + market_data = "\n\n".join([ + f"Market Report:\n{self.curr_state.get('market_report', '')}", + f"Sentiment Report:\n{self.curr_state.get('sentiment_report', '')}", + f"News Report:\n{self.curr_state.get('news_report', '')}", + f"Fundamentals Report:\n{self.curr_state.get('fundamentals_report', '')}", + ]) + + decision = self.curr_state.get("final_trade_decision", "") + + self.ace_engine.learn_from_trade( + context=context, + decision=decision, + result=str(returns_losses), + market_data=market_data, + ) + + def get_ace_context(self) -> str: + """Get ACE strategies context for injection into agent prompts.""" + if self.ace_enabled and self.ace_engine: + return self.ace_engine.get_skills_context() + return "" + + def save_ace_skillbook(self, path: Optional[str] = None) -> str: + """Save the ACE skillbook to a file.""" + if self.ace_engine: + return self.ace_engine.save_skillbook(path) + return "" + + def get_ace_stats(self) -> Dict[str, Any]: + """Get ACE learning statistics.""" + if self.ace_engine: + return self.ace_engine.get_stats() + return {} + + def _ace_learn_from_analysis(self): + """ + Trigger ACE learning based on the analytical consistency of all reports. + """ + if not self.ace_enabled or not self.ace_engine or not self.curr_state: + return + + print(f"DEBUG: ACE analytical reflection triggered for {self.curr_state.get('company_of_interest')}") + + reports = { + "ticker": self.curr_state.get("company_of_interest", "Unknown"), + "date": str(self.curr_state.get("trade_date", "Unknown")), + "market": self.curr_state.get("market_report", ""), + "sentiment": self.curr_state.get("sentiment_report", ""), + "news": self.curr_state.get("news_report", ""), + "fundamentals": self.curr_state.get("fundamentals_report", ""), + "plan": self.curr_state.get("investment_plan", ""), + } + + decision = self.curr_state.get("final_trade_decision", "") + # Clean metadata tag if present + decision = re.sub(r"\[ACE_METADATA: .*\]", "", decision).strip() + + self.ace_engine.learn_from_analysis( + reports=reports, + decision=decision + ) + + def get_ace_context(self) -> str: + """Get ACE strategies context for injection into agent prompts.""" + if self.ace_enabled and self.ace_engine: + return self.ace_engine.get_skills_context() + return "" + + def save_ace_skillbook(self, path: Optional[str] = None) -> str: + """Save the ACE skillbook to a file.""" + if self.ace_engine: + return self.ace_engine.save_skillbook(path) + return "" + + def get_ace_stats(self) -> Dict[str, Any]: + """Get ACE learning statistics.""" + if self.ace_engine: + return self.ace_engine.get_stats() + return {} + def process_signal(self, full_signal): """Process a signal to extract the core decision.""" return self.signal_processor.process_signal(full_signal)