trending/AutoHedge
1
0
Fork 0
mirror of https://github.com/The-Swarm-Corporation/AutoHedge.git synced 2026-10-02 05:24:49 +08:00
Code Issues Packages Projects Releases Wiki Activity

autohedge improvement and cli

Browse Source
This commit is contained in:
Kye Gomez
2026-02-17 14:30:57 -08:00
parent a24c2dbb27
commit 3346a0edb0
10 changed files with 210 additions and 811 deletions
Download Patch File Download Diff File Expand all files Collapse all files
README.md
+11 -8
View File
@@ -52,8 +52,17 @@ pip install -U autohedge
### Environment Variables
```bash
OPENAI_API_KEY=""
# Jupiter API (token price & search tools)
# Get a key at https://portal.jup.ag
JUPITER_API_KEY=
# OpenAI (experimental agents)
OPENAI_API_KEY=
ANTHROPIC_API_KEY=
WORKSPACE_DIR="agent_workspace"
# Trading
WALLET_PRIVATE_KEY=""
```
See `.env.example` for a full reference.
@@ -61,13 +70,7 @@ See `.env.example` for a full reference.
### Basic Usage
```python
from autohedge import AutoHedge
stocks = ["NVDA"]
trading_system = AutoHedge(stocks)
task = "Analyze NVIDIA for a 50k allocation and recommend action."
print(trading_system.run(task=task))
autohedge
```
---
autohedge/cli.py
+9 -31
View File
@@ -2,7 +2,6 @@
AutoHedge CLI — welcome screen and interactive REPL.
"""
import os
import sys
from pathlib import Path
@@ -11,6 +10,9 @@ from rich.panel import Panel
from rich.text import Text
from rich.columns import Columns
from rich import box
from dotenv import load_dotenv
load_dotenv()
try:
from importlib.metadata import version as _version
@@ -31,8 +33,7 @@ BANNER_ART = r"""
"""
TIPS = [
"Enter a task to run (e.g. 'Analyze NVDA for 50k allocation')",
"Type 'stocks AAPL,MSFT' to set tickers, then run a task",
"Enter a task prompt to run (e.g. 'Analyze NVDA for 50k allocation')",
"Type 'quit' or 'exit' to leave",
"Type 'help' or '?' for commands",
]
@@ -82,8 +83,7 @@ def _welcome() -> None:
tips_text = Text(
"Tips for getting started\n", style="bold orange1"
)
tips_text.append(" — ".join(TIPS[:2]) + "\n", style="dim")
tips_text.append(" — ".join(TIPS[2:]), style="dim")
tips_text.append(" — ".join(TIPS), style="dim")
recent = _get_recent_tasks()
recent_heading = Text("Recent activity\n", style="bold orange1")
@@ -135,8 +135,7 @@ def _welcome() -> None:
)
def run_repl(default_stocks: list[str] | None = None) -> None:
stocks = default_stocks or ["NVDA"]
def run_repl() -> None:
_welcome()
while True:
@@ -161,25 +160,13 @@ def run_repl(default_stocks: list[str] | None = None) -> None:
console.print(f" [dim]·[/] {t}")
continue
if lower.startswith("stocks "):
raw = line[6:].strip()
stocks = [
s.strip().upper()
for s in raw.replace(",", " ").split()
if s.strip()
]
console.print(
f"[dim]Stocks set to: {', '.join(stocks)}[/]"
)
continue
# Treat as task
# Treat as task prompt
task = line
_append_recent(task)
try:
from autohedge import AutoHedge
system = AutoHedge(stocks=stocks)
system = AutoHedge()
console.print("[dim]Running...[/]")
result = system.run(task=task)
console.print(
@@ -195,16 +182,7 @@ def run_repl(default_stocks: list[str] | None = None) -> None:
def main() -> None:
"""Entry point for the AutoHedge CLI."""
default_stocks = os.environ.get("AUTOHEDGE_STOCKS")
if default_stocks:
stocks = [
s.strip().upper()
for s in default_stocks.split(",")
if s.strip()
]
else:
stocks = None
run_repl(default_stocks=stocks)
run_repl()
sys.exit(0)
autohedge/main.py
+10 -146
View File
@@ -1,93 +1,34 @@
import uuid
from datetime import datetime
from pathlib import Path
from typing import List, Optional
from loguru import logger
from pydantic import BaseModel
from swarms import Conversation
from autohedge.workers import (
ExecutionAgent,
QuantAnalyst,
RiskManager,
TradingDirector,
)
from autohedge.workers import director_agent
class AutoHedgeOutput(BaseModel):
id: str = uuid.uuid4().hex
thesis: Optional[str] = None
risk_assessment: Optional[str] = None
order: Optional[str] = None
decision: str = None
timestamp: str = datetime.now().isoformat()
current_stock: str
class AutoHedgeOutputMain(BaseModel):
name: Optional[str] = None
description: Optional[str] = None
id: str = uuid.uuid4().hex
stocks: Optional[list] = None
task: Optional[str] = None
timestamp: str = datetime.now().isoformat()
logs: List[AutoHedgeOutput] = None
class AutoHedge:
"""
Main trading system that coordinates all agents and manages the trading cycle.
Attributes:
stocks (List[str]): List of stock tickers to trade
director (TradingDirector): Trading director agent
quant (QuantAnalyst): Quantitative analysis agent
risk (RiskManager): Risk management agent
execution (ExecutionAgent): Trade execution agent
output_dir (Path): Directory for storing outputs
Tickers to analyze are derived from the task by the director (no predefined list).
"""
def __init__(
self,
stocks: List[str],
name: str = "autohedge",
description: str = "fully autonomous hedgefund",
output_dir: str = "outputs",
output_file_path: str = None,
strategy: str = None,
output_type: str = "list",
):
"""
Initialize the AutoHedge class.
Args:
stocks (List[str]): List of stock tickers to trade
name (str, optional): Name of the trading system. Defaults to "autohedge".
description (str, optional): Description of the trading system. Defaults to "fully autonomous hedgefund".
output_dir (str, optional): Directory for storing outputs. Defaults to "outputs".
output_file_path (str, optional): Path to the output file. Defaults to None.
"""
self.name = name
self.description = description
self.stocks = stocks
self.output_dir = Path(output_dir)
self.output_dir.mkdir(exist_ok=True)
self.strategy = strategy
self.output_type = output_type
self.output_file_path = output_file_path
self.output_dir = Path(output_dir)
self.output_dir.mkdir(exist_ok=True)
logger.info("Initializing Automated Trading System")
self.director = TradingDirector(stocks, output_dir)
self.quant = QuantAnalyst()
self.risk = RiskManager()
self.execution = ExecutionAgent()
self.logs = AutoHedgeOutputMain(
name=self.name,
description=self.description,
stocks=stocks,
task="",
logs=[],
)
self.conversation = Conversation(time_enabled=True)
def run(self, task: str, *args, **kwargs):
@@ -106,95 +47,18 @@ class AutoHedge:
self.conversation.add(role="user", content=f"Task: {task}")
try:
for stock in self.stocks:
logger.info(f"Processing {stock}")
output = director_agent.run(task=task)
self.conversation.add(role="director", content=output)
# Generate thesis
thesis, market_data = self.director.generate_thesis(
task=task, stock=stock
)
self.conversation.add(
role=self.director.agent_name,
content=f"Stock: {stock}\nMarket Data: {market_data}\nThesis: {thesis}",
)
# Perform analysis
analysis = self.quant.analyze(
stock + market_data, thesis, task=task
)
# setiment_analysis = sentiment_agent.run(
# fetch_stock_news(stock)
# )
# logger.info(f"Sentiment Analysis: {setiment_analysis}")
# self.conversation.add(sentiment_agent.agent_name, setiment_analysis)
self.conversation.add(
role=self.quant.agent_name, content=analysis
)
# Assess risk
risk_assessment = self.risk.assess_risk(
stock + market_data, thesis, analysis, task=task
)
self.conversation.add(
role=self.risk.agent_name, content=risk_assessment
)
# # Generate order if approved
order = self.execution.generate_order(
stock, thesis, risk_assessment, task=task
)
self.conversation.add(
role=self.execution.agent_name, content=order
)
order = str(order)
# Final decision
decision = self.director.make_decision(
order + market_data + risk_assessment,
thesis,
user_task=task,
)
self.conversation.add(
role=self.director.agent_name, content=decision
)
# log = AutoHedgeOutput(
# thesis=thesis,
# risk_assessment=risk_assessment,
# current_stock=stock,
# order=order,
# decision=decision,
# )
# # logs.append(log.model_dump_json(indent=4))
# self.logs.task = task
# self.logs.logs.append(log)
# create_file_in_folder(
# self.output_dir,
# f"analysis-{uuid.uuid4().hex}.json",
# self.logs.model_dump_json(indent=4),
# )
# return self.logs.model_dump_json(indent=4)
if self.output_type == "list":
return self.conversation.return_messages_as_list()
elif self.output_type == "dict":
if self.output_type == "dict":
return (
self.conversation.return_messages_as_dictionary()
)
elif self.output_type == "str":
if self.output_type == "str":
return self.conversation.return_history_as_string()
return self.conversation.return_messages_as_list()
except Exception as e:
logger.error(f"Error in trading cycle: {str(e)}")
raise
autohedge/prompts.py
+9
View File
@@ -191,3 +191,12 @@ Generate quantitative analysis for the {stock}
"""
DIRECTOR_DECISION_PROMPT = "According to the thesis, {thesis}, should we execute this order: {task}"
# Director: discover tickers from task (no predefined list)
DIRECTOR_TICKER_DISCOVERY_PROMPT = """
Given the following task, determine which stock tickers are relevant to analyze.
Task: {task}
Reply with ONLY a JSON array of ticker symbols (e.g. ["NVDA", "MSFT", "GOOG"]). Use US exchange symbols. No other text.
"""
autohedge/tools/e_trade_wrapper.py
-175
View File
@@ -1,175 +0,0 @@
import os
from typing import Dict, Any, Union
from loguru import logger
from dotenv import load_dotenv
from requests_oauthlib import OAuth1Session
# Load environment variables from .env file
load_dotenv()
class ETradeClient:
"""
A client for interacting with the E*TRADE API to manage trades and accounts.
"""
# BASE_URL = "https://api.etrade.com/v1" # Sandbox base URL
# For production, replace with "https://api.etrade.com/v1"
BASE_URL = "https://api.etrade.com/v1"
def __init__(self, account_id: str, production_url: str):
"""
Initialize the E*TRADE client with OAuth credentials from environment variables.
"""
self.consumer_key = os.getenv("ETRADE_CONSUMER_KEY")
self.consumer_secret = os.getenv("ETRADE_CONSUMER_SECRET")
self.oauth_token = os.getenv("ETRADE_OAUTH_TOKEN")
self.oauth_token_secret = os.getenv(
"ETRADE_OAUTH_TOKEN_SECRET"
)
self.account_id = account_id
if not all(
[
self.consumer_key,
self.consumer_secret,
self.oauth_token,
self.oauth_token_secret,
]
):
logger.error(
"E*TRADE credentials are not set in the environment variables."
)
raise EnvironmentError("Missing E*TRADE credentials.")
self.oauth_session = OAuth1Session(
client_key=self.consumer_key,
client_secret=self.consumer_secret,
resource_owner_key=self.oauth_token,
resource_owner_secret=self.oauth_token_secret,
)
logger.success("Initialized E*TRADE client.")
def place_order(
self,
account_id: str,
symbol: str,
quantity: int,
action: str,
price: Union[float, None] = None,
) -> Dict[str, Any]:
"""
Place a buy or sell order.
Args:
account_id: The account ID for placing the order.
symbol: The stock ticker symbol (e.g., 'AAPL').
quantity: Number of shares.
action: 'BUY' or 'SELL'.
price: Limit price (optional, for limit orders).
Returns:
Response JSON from the API.
"""
url = f"{self.BASE_URL}/accounts/{account_id}/orders/place"
order_payload = {
"orderType": "LIMIT" if price else "MARKET",
"clientOrderId": (
"12345"
), # Replace with dynamic unique ID in production
"orderAction": action.upper(),
"instrument": [
{
"symbol": symbol,
"quantity": quantity,
"orderAction": action.upper(),
}
],
"priceType": "LIMIT" if price else "MARKET",
"limitPrice": price if price else None,
"marketSession": "REGULAR",
"orderTerm": "GOOD_FOR_DAY",
}
try:
logger.info(
f"Placing {action.upper()} order for {quantity} shares of {symbol} (Limit: {price})"
)
response = self.oauth_session.post(
url, json=order_payload
)
response.raise_for_status()
logger.success("Order placed successfully.")
return response.json()
except Exception as e:
logger.error(f"Failed to place order: {e}")
raise
def get_account_info(self) -> Dict[str, Any]:
"""
Fetch account information, including balances and positions.
Args:
account_id: The account ID.
Returns:
A dictionary containing account details.
"""
url = f"{self.BASE_URL}/accounts/{self.account_id}/balance"
try:
logger.info("Fetching account information...")
response = self.oauth_session.get(url)
response.raise_for_status()
logger.success(
"Fetched account information successfully."
)
return response.json()
except Exception as e:
logger.error(f"Failed to fetch account information: {e}")
raise
def logout(self) -> None:
"""
End the session with E*TRADE.
"""
try:
logger.info("Ending E*TRADE session...")
# E*TRADE does not require explicit logout; session ends automatically.
logger.success("Session ended.")
except Exception as e:
logger.error(f"Failed to end session: {e}")
raise
def get_acc_info():
client = ETradeClient(account_id=os.getenv("ETRADE_ACCOUNT_ID"))
return client.get_account_info()
# # Example Usage
# if __name__ == "__main__":
# logger.add("etrade_client.log", rotation="1 MB", retention="10 days", level="DEBUG")
# try:
# client = ETradeClient()
# # Fetch account info
# account_id = "12345678" # Replace with your account ID
# account_info = client.get_account_info(account_id)
# logger.info(f"Account Info: {account_info}")
# # Place a buy order
# buy_response = client.place_order(account_id, symbol="AAPL", quantity=10, action="BUY", price=150.00)
# logger.info(f"Buy Order Response: {buy_response}")
# # Place a sell order
# sell_response = client.place_order(account_id, symbol="AAPL", quantity=10, action="SELL")
# logger.info(f"Sell Order Response: {sell_response}")
# except Exception as e:
# logger.error(f"An error occurred: {e}")
# finally:
# # Logout not required for E*TRADE
# logger.info("Execution completed.")
autohedge/tools/exa_search_tool.py
+104
View File
@@ -0,0 +1,104 @@
import os
import httpx
from loguru import logger
from swarms.utils.any_to_str import any_to_str
def exa_search(
query: str,
) -> str:
"""
Exa Web Search Tool
This function provides advanced, natural language web search capabilities
using the Exa.ai API. It is designed for use by research agents and
subagents to retrieve up-to-date, relevant information from the web,
including documentation, technical articles, and general knowledge sources.
Features:
- Accepts natural language queries (e.g., "Find the latest PyTorch 2.2.0 documentation on quantization APIs")
- Returns structured, summarized results suitable for automated research workflows
- Supports parallel execution for multiple subagents
- Can be used to search for:
* Official documentation (e.g., Python, PyTorch, TensorFlow, API docs)
* Research papers and technical blogs
* News, regulatory updates, and more
Args:
query (str): The natural language search query. Can be a question, a request for documentation, or a technical prompt.
Returns:
str: JSON-formatted string containing the search results, including summaries and key insights.
Example usage:
exa_search("Show me the latest Python 3.12 documentation on dataclasses")
exa_search("Recent research on transformer architectures for vision tasks")
Notes:
- This tool is ideal for agents that need to quickly gather authoritative information from the web, especially official docs.
- The Exa API is capable of extracting and summarizing content from a wide range of sources, including documentation sites, arXiv, blogs, and more.
- For best results when searching for documentation, include the technology/library name and the specific topic or API in your query.
"""
api_key = os.getenv("EXA_API_KEY")
if not api_key:
raise ValueError(
"EXA_API_KEY environment variable is not set"
)
characters = 20
sources = 2
headers = {
"x-api-key": api_key,
"content-type": "application/json",
}
# Payload format for Exa API (see https://docs.exa.ai/reference/search)
payload = {
"query": query,
"type": "auto",
"numResults": sources,
"contents": {
"text": True,
"summary": {
"schema": {
"type": "object",
"required": ["answer"],
"additionalProperties": False,
"properties": {
"answer": {
"type": "string",
"description": (
"Key insights and findings from the search result"
),
}
},
}
},
"context": {"maxCharacters": characters},
},
}
try:
logger.info(
f"[SEARCH] Executing Exa search for: {query[:50]}..."
)
response = httpx.post(
"https://api.exa.ai/search",
json=payload,
headers=headers,
timeout=30,
)
response.raise_for_status()
json_data = response.json()
return any_to_str(json_data)
except Exception as e:
logger.error(f"Exa search failed: {e}")
return f"Search failed: {str(e)}. Please try again."
autohedge/tools/polygon_api.py
+3 -4
View File
@@ -1,6 +1,6 @@
"""
Stocks API client for ticker overview, balance sheets, and daily OHLC.
Uses Massive API (https://massive.com/docs). Set POLYGON_API_KEY and optionally
Uses Massive API (https://massive.com/docs). Set MASSIVE_API_KEY and optionally
POLYGON_BASE_URL in .env.
"""
@@ -12,12 +12,11 @@ import httpx
from loguru import logger
DEFAULT_BASE_URL = "https://api.massive.com"
BASE_URL = os.getenv("POLYGON_BASE_URL", DEFAULT_BASE_URL)
def _get_headers() -> dict[str, str]:
headers: dict[str, str] = {}
key = os.getenv("POLYGON_API_KEY")
key = os.getenv("MASSIVE_API_KEY")
if key:
headers["Authorization"] = f"Bearer {key}"
return headers
@@ -28,7 +27,7 @@ def _get(
*,
params: Optional[dict[str, Any]] = None,
) -> str:
url = f"{BASE_URL.rstrip('/')}{path}"
url = f"{DEFAULT_BASE_URL.rstrip('/')}{path}"
try:
with httpx.Client(timeout=15) as client:
resp = client.get(
autohedge/workers.py
+61 -436
View File
@@ -1,35 +1,63 @@
"""
AutoHedge workers: Pydantic output models and agent classes for thesis
AutoHedge workers: Pydantic output models and agents for thesis
generation, risk assessment, execution, and quantitative analysis.
"""
import uuid
from datetime import datetime
from pathlib import Path
from typing import Dict, List, Optional
from loguru import logger
from pydantic import BaseModel
from swarms import Agent
from autohedge.prompts import (
DIRECTOR_DECISION_PROMPT,
DIRECTOR_PROMPT,
DIRECTOR_THESIS_PROMPT,
EXECUTION_ORDER_PROMPT,
EXECUTION_PROMPT,
QUANT_ANALYSIS_PROMPT,
QUANT_PROMPT,
RISK_ASSESSMENT_PROMPT,
RISK_PROMPT,
SENTIMENT_PROMPT,
)
from autohedge.tools.yahoo_api import get_all_stock_data
from autohedge.tools.exa_search_tool import exa_search
_NOW = datetime.now().isoformat()
_SYSTEM_SUFFIX = f"\n\nCurrent date and time: {_NOW}"
sentiment_agent = Agent(
agent_name="Sentiment-Agent",
system_prompt=SENTIMENT_PROMPT,
system_prompt=SENTIMENT_PROMPT + _SYSTEM_SUFFIX,
model_name="gpt-4o-mini",
verbose=True,
max_loops=1,
tools=[exa_search],
)
risk_agent = Agent(
agent_name="Risk-Manager",
system_prompt=RISK_PROMPT.strip()
+ "\n\nWhen you receive a message, it will contain:\nStock, Thesis, Quant Analysis.\n\nProvide risk assessment including:\n1. Recommended position size\n2. Maximum drawdown risk\n3. Market risk exposure\n4. Overall risk score"
+ _SYSTEM_SUFFIX,
model_name="gpt-4.1",
output_type="str",
max_loops=1,
verbose=True,
context_length=16000,
)
execution_agent = Agent(
agent_name="Execution-Agent",
system_prompt=EXECUTION_PROMPT.strip()
+ "\n\nWhen you receive a message, it will contain:\nStock, Thesis, Risk Assessment.\n\nGenerate trade order including:\n1. Order type (market/limit)\n2. Quantity\n3. Entry price\n4. Stop loss\n5. Take profit\n6. Time in force"
+ _SYSTEM_SUFFIX,
model_name="gpt-4.1",
output_type="str",
max_loops=1,
verbose=True,
context_length=16000,
)
quant_agent = Agent(
agent_name="Quant-Analyst",
system_prompt=QUANT_PROMPT.strip()
+ "\n\nWhen you receive a message, it will contain:\nStock and Thesis from your Director.\n\nGenerate quantitative analysis with: ticker, technical_score (0-1), volume_score (0-1), trend_strength (0-1), volatility, probability_score (0-1), key_levels (support, resistance, pivot)."
+ _SYSTEM_SUFFIX,
model_name="gpt-4.1",
output_type="str",
max_loops=1,
verbose=True,
@@ -37,428 +65,25 @@ sentiment_agent = Agent(
)
def _agent_context(task: Optional[str] = None) -> str:
"""Build context string with current time and task to prepend to agent prompts."""
now = datetime.now().isoformat()
task_str = task if task else "(none)"
return f"Current time: {now}\nTask: {task_str}\n\n"
class AutoHedgeOutput(BaseModel):
"""
Per-stock output from the AutoHedge pipeline for a single ticker.
Attributes
----------
id : str
Unique run identifier (hex UUID).
thesis : str, optional
Trading thesis for the stock.
risk_assessment : str, optional
Risk assessment text from the risk manager.
order : str, optional
Generated order / execution plan.
decision : str, optional
Director decision (e.g. hold, buy, sell).
timestamp : str
ISO timestamp when this output was produced.
current_stock : str
Ticker symbol this output refers to.
"""
id: str = uuid.uuid4().hex
thesis: Optional[str] = None
risk_assessment: Optional[str] = None
order: Optional[str] = None
decision: str = None
timestamp: str = datetime.now().isoformat()
current_stock: str
class AutoHedgeOutputMain(BaseModel):
"""
Top-level output from a full AutoHedge run (multiple stocks / task).
Attributes
----------
name : str, optional
Run or strategy name.
description : str, optional
Human-readable description of the run.
id : str
Unique run identifier (hex UUID).
stocks : list, optional
List of ticker symbols processed.
task : str, optional
User task or instruction for the run.
timestamp : str
ISO timestamp when the run completed.
logs : list of AutoHedgeOutput, optional
Per-stock results in order of processing.
"""
name: Optional[str] = None
description: Optional[str] = None
id: str = uuid.uuid4().hex
stocks: Optional[list] = None
task: Optional[str] = None
timestamp: str = datetime.now().isoformat()
logs: List[AutoHedgeOutput] = None
class RiskManager:
"""
Agent that assesses risk for a stock given a thesis and quant analysis.
Uses a dedicated Swarms agent (RISK_PROMPT) to produce a text risk
assessment from the trading thesis and quantitative analysis.
Attributes
----------
risk_agent : Agent
Swarms agent used for risk assessment.
"""
def __init__(self):
self.risk_agent = Agent(
agent_name="Risk-Manager",
system_prompt=RISK_PROMPT,
model_name="gpt-4.1",
output_type="str",
max_loops=1,
verbose=True,
context_length=16000,
)
def assess_risk(
self,
stock: str,
thesis: str,
quant_analysis: str,
task: Optional[str] = None,
) -> str:
"""
Produce a risk assessment for a stock given thesis and quant analysis.
Parameters
----------
stock : str
Ticker symbol.
thesis : str
Trading thesis text.
quant_analysis : str
Quantitative analysis text.
task : str, optional
User task or instruction (included in agent context with current time).
Returns
-------
str
Risk assessment text from the risk agent.
"""
prompt = _agent_context(task) + RISK_ASSESSMENT_PROMPT.format(
stock=stock, thesis=thesis, quant_analysis=quant_analysis
)
assessment = self.risk_agent.run(prompt)
return assessment
class ExecutionAgent:
"""
Agent that generates execution orders from thesis and risk assessment.
Uses a Swarms agent (EXECUTION_PROMPT) to turn a thesis and risk
assessment into a concrete order or execution plan.
Attributes
----------
execution_agent : Agent
Swarms agent used for order generation.
"""
def __init__(self):
self.execution_agent = Agent(
agent_name="Execution-Agent",
system_prompt=EXECUTION_PROMPT,
model_name="gpt-4.1",
output_type="str",
max_loops=1,
verbose=True,
context_length=16000,
)
def generate_order(
self,
stock: str,
thesis: Dict,
risk_assessment: Dict,
task: Optional[str] = None,
) -> str:
"""
Generate an execution order for a stock from thesis and risk data.
Parameters
----------
stock : str
Ticker symbol.
thesis : dict
Trading thesis (or serialized thesis data).
risk_assessment : dict
Risk assessment (or serialized risk data).
task : str, optional
User task or instruction (included in agent context with current time).
Returns
-------
str
Generated order text from the execution agent.
"""
prompt = _agent_context(task) + EXECUTION_ORDER_PROMPT.format(
stock=stock,
thesis=thesis,
risk_assessment=risk_assessment,
)
order = self.execution_agent.run(prompt)
return order
class TradingDirector:
"""
Coordinates strategy and generates trading theses using market data.
Uses a Swarms director agent and Yahoo Finance data to fetch market data,
then produces a trading thesis and can make follow-up decisions.
Attributes
----------
director_agent : Agent
Swarms agent for thesis generation and decisions.
"""
def __init__(
self,
stocks: List[str],
output_dir: str = "outputs",
cryptos: List[str] = None,
):
"""
Parameters
----------
stocks : list of str
Ticker symbols the director may analyze.
output_dir : str, optional
Directory for outputs (default "outputs").
cryptos : list of str, optional
Crypto symbols for crypto thesis (currently unused).
"""
logger.info("Initializing Trading Director")
self.director_agent = Agent(
agent_name="Trading-Director",
system_prompt=DIRECTOR_PROMPT,
model_name="gpt-4.1",
output_type="str",
max_loops=1,
verbose=True,
context_length=16000,
)
# self.crypto_agent = CryptoAgentWrapper()
def generate_thesis(
self,
task: str = "Generate a thesis for the stock",
stock: str = None,
crypto: str = None,
) -> str:
"""
Generate trading thesis for a given stock.
Args:
stock (str): Stock ticker symbol
Returns:
TradingThesis: Generated thesis
"""
logger.info(f"Generating thesis for {stock}")
try:
market_data = get_all_stock_data(
stock, include_history=True
)
prompt = _agent_context(
task
) + DIRECTOR_THESIS_PROMPT.format(
task=task, stock=stock, market_data=market_data
)
thesis = self.director_agent.run(prompt)
return thesis, market_data
except Exception as e:
logger.error(
f"Error generating thesis for {stock}: {str(e)}"
)
raise
def make_decision(
self,
task: str,
thesis: str,
user_task: Optional[str] = None,
*args,
**kwargs,
):
"""
Run the director agent to make a decision given a task and thesis.
Parameters
----------
task : str
Order/context to evaluate (e.g. order + market_data + risk_assessment).
thesis : str
Trading thesis text.
user_task : str, optional
User task or instruction (included in agent context with current time).
*args, **kwargs
Passed through to the agent (e.g. for future options).
Returns
-------
str
Director decision output.
"""
prompt = _agent_context(
user_task
) + DIRECTOR_DECISION_PROMPT.format(thesis=thesis, task=task)
return self.director_agent.run(prompt)
def generate_thesis_crypto(
self,
task: str = None,
crypto: str = None,
):
"""
Generate a trading thesis for a crypto asset using the crypto agent.
Parameters
----------
task : str, optional
Analysis task or instruction.
crypto : str, optional
Crypto symbol (e.g. BTC, ETH).
Returns
-------
str
Generated thesis text.
Raises
------
Exception
If crypto_agent is not set or the run fails.
"""
logger.info(f"Generating thesis for {crypto}")
try:
market_data = self.crypto_agent.run(
crypto,
f"{task} Analyze current market conditions and key metrics for {crypto}",
)
prompt = _agent_context(
task
) + DIRECTOR_THESIS_PROMPT.format(
task=task, stock=crypto, market_data=market_data
)
thesis = self.director_agent.run(prompt)
return thesis
except Exception as e:
logger.error(
f"Error generating thesis for {crypto}: {str(e)}"
)
raise
class QuantAnalyst:
"""
Agent that performs quantitative (technical and statistical) analysis.
Uses a Swarms agent (QUANT_PROMPT) to analyze a stock in the context
of a trading thesis and produce structured quant analysis.
Attributes
----------
quant_agent : Agent
Swarms agent used for analysis.
output_dir : Path
Directory for saving outputs (created on init).
"""
def __init__(self, output_dir: str = "outputs"):
self.output_dir = Path(output_dir)
self.output_dir.mkdir(exist_ok=True)
logger.info("Initializing Quant Analyst")
self.quant_agent = Agent(
agent_name="Quant-Analyst",
system_prompt=QUANT_PROMPT,
model_name="gpt-4.1",
output_type="str",
max_loops=1,
verbose=True,
context_length=16000,
)
def analyze(
self, stock: str, thesis: str, task: Optional[str] = None
) -> str:
"""
Perform quantitative analysis for a stock given a trading thesis.
Parameters
----------
stock : str
Stock ticker symbol.
thesis : str
Trading thesis text.
task : str, optional
User task or instruction (included in agent context with current time).
Returns
-------
str
Quantitative analysis text from the quant agent.
"""
logger.info(f"Performing quant analysis for {stock}")
try:
prompt = _agent_context(
task
) + QUANT_ANALYSIS_PROMPT.format(
stock=stock, thesis=thesis
)
analysis = self.quant_agent.run(prompt)
return analysis
except Exception as e:
logger.error(
f"Error in quant analysis for {stock}: {str(e)}"
)
raise
# -----------------------------------------------------------------------------
# Initialized workers and their agents (for discovery / iteration)
# -----------------------------------------------------------------------------
risk_manager = RiskManager()
execution_agent_instance = ExecutionAgent()
trading_director = TradingDirector(stocks=[])
quant_analyst = QuantAnalyst()
ALL_AGENTS = [
sentiment_agent, # Sentiment-Agent
risk_manager.risk_agent, # Risk-Manager
execution_agent_instance.execution_agent, # Execution-Agent
trading_director.director_agent, # Trading-Director
quant_analyst.quant_agent, # Quant-Analyst
sentiment_agent,
risk_agent,
execution_agent,
quant_agent,
]
director_agent = Agent(
agent_name="Trading-Director",
system_prompt=DIRECTOR_PROMPT + _SYSTEM_SUFFIX,
model_name="gpt-4.1",
max_loops=1,
handoffs=ALL_AGENTS,
)
if __name__ == "__main__":
output = director_agent.run(
"Analyze the stock market and provide a thesis on the overall market position and expected trends."
)
print(output)
example.py
+2 -10
View File
@@ -3,19 +3,11 @@ from autohedge.main import AutoHedge
load_dotenv()
# Define the stocks to analyze
stocks = ["NVDA", "TSLA", "MSFT", "GOOG"]
# Initialize the trading system with the specified stocks
# Initialize the trading system (tickers are derived from the task by the director)
trading_system = AutoHedge(
name="swarms-fund",
description="Private Hedge Fund for Swarms Corp",
stocks=stocks,
)
# Define the task for the trading cycle
task = "As BlackRock, let's evaluate AI companies for a portfolio with $500 million in allocation, aiming for a balanced risk-reward profile."
# Run the trading cycle and print the results
task = "Analyze the sentiment of oil market and provide a thesis on the overall market position and expected trends."
print(trading_system.run(task=task))
pyproject.toml
+1 -1
View File
@@ -4,7 +4,7 @@ build-backend = "poetry.core.masonry.api"
[tool.poetry]
name = "autohedge"
version = "0.1.2"
version = "0.1.3"
description = "autohedge - TGSC"
license = "MIT"
authors = ["Kye Gomez <kye@apac.ai>"]