Spaces:
Sleeping
Sleeping
File size: 4,394 Bytes
68025ee | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 | import logging
from agents.trader import Trader
from agents.analysts import TechnicalAnalyst, NewsAnalyst, SentimentAnalyst
from agents.researcher import Researcher
from agents.risk_manager import RiskManager
from llm.openrouter import OpenRouterClient
logger = logging.getLogger(__name__)
class SingleAgentPipeline:
"""Benchmark A: single Trader agent with price + indicators."""
def __init__(self, trader: Trader):
self.trader = trader
def decide(self, market_data: dict) -> dict:
decision = self.trader.run(market_data)
return {
"decision": decision,
"agent_outputs": {"trader": decision},
}
class SimplePipeline:
"""Benchmark B: TechnicalAnalyst + NewsAnalyst -> Trader."""
def __init__(self, analysts: list, trader: Trader):
self.technical = next(a for a in analysts if isinstance(a, TechnicalAnalyst))
self.news = next(a for a in analysts if isinstance(a, NewsAnalyst))
self.trader = trader
def decide(self, market_data: dict) -> dict:
tech_analysis = self.technical.run(market_data)
news_analysis = self.news.run(market_data)
trader_context = {
**market_data,
"tech_analysis": tech_analysis,
"news_analysis": news_analysis,
}
decision = self.trader.run(trader_context)
return {
"decision": decision,
"agent_outputs": {
"technical_analyst": tech_analysis,
"news_analyst": news_analysis,
"trader": decision,
},
}
class FullPipeline:
"""Benchmark C: Technical + Sentiment + News -> Researcher -> RiskManager -> Trader."""
def __init__(self, analysts: list, researcher: Researcher, risk_manager: RiskManager, trader: Trader):
self.technical = next(a for a in analysts if isinstance(a, TechnicalAnalyst))
self.sentiment = next(a for a in analysts if isinstance(a, SentimentAnalyst))
self.news = next(a for a in analysts if isinstance(a, NewsAnalyst))
self.researcher = researcher
self.risk_manager = risk_manager
self.trader = trader
def decide(self, market_data: dict) -> dict:
# Phase 1: analysts
tech_analysis = self.technical.run(market_data)
news_analysis = self.news.run(market_data)
sentiment_analysis = self.sentiment.run(market_data)
# Phase 2: researcher bull/bear debate
research_context = {
**market_data,
"tech_analysis": tech_analysis,
"news_analysis": news_analysis,
"sentiment_analysis": sentiment_analysis,
}
research = self.researcher.run(research_context)
# Phase 3: risk manager
portfolio = market_data.get("portfolio", {})
risk_context = {
"recommendation": research,
"portfolio": portfolio,
}
risk_decision = self.risk_manager.run(risk_context)
# Phase 4: final trader decision
trader_context = {
**market_data,
"research": research,
"risk_decision": risk_decision,
}
decision = self.trader.run(trader_context)
return {
"decision": decision,
"agent_outputs": {
"technical_analyst": tech_analysis,
"news_analyst": news_analysis,
"sentiment_analyst": sentiment_analysis,
"researcher": research,
"risk_manager": risk_decision,
"trader": decision,
},
}
def build_pipeline(benchmark: str, model: str):
"""Factory: build the correct pipeline for benchmark A/B/C."""
llm = OpenRouterClient(model=model)
if benchmark == "A":
return SingleAgentPipeline(trader=Trader(llm, benchmark="A"))
if benchmark == "B":
return SimplePipeline(
analysts=[TechnicalAnalyst(llm), NewsAnalyst(llm)],
trader=Trader(llm, benchmark="B"),
)
if benchmark == "C":
return FullPipeline(
analysts=[TechnicalAnalyst(llm), SentimentAnalyst(llm), NewsAnalyst(llm)],
researcher=Researcher(llm),
risk_manager=RiskManager(llm),
trader=Trader(llm, benchmark="C"),
)
raise ValueError(f"Unknown benchmark: {benchmark}")
|