Define Crypto Tools with Atomic Schema
from atomic_agents.tools.base_tool import BaseTool, BaseToolConfig
from atomic_agents.lib.base.base_io_schema import BaseIOSchema
import purpleflea as pf
from pydantic import Field
from typing import Literal
pf_client = pf.Client(api_key="YOUR_KEY")
class GetPriceInput(BaseIOSchema):
symbol: str = Field(description="Crypto symbol e.g. BTC, ETH")
class GetPriceOutput(BaseIOSchema):
symbol: str
price: float
change_24h: float
funding_rate: float
signal: Literal["bullish", "bearish", "neutral"]
class OpenTradeInput(BaseIOSchema):
symbol: str = Field(description="Market to trade")
side: Literal["long", "short"]
size_usd: float = Field(gt=0, le=500)
leverage: int = Field(1, ge=1, le=10)
class OpenTradeOutput(BaseIOSchema):
position_id: str
symbol: str
side: str
fill_price: float
liquidation_price: float
estimated_pnl: float
class GetPriceTool(BaseTool):
input_schema = GetPriceInput
output_schema = GetPriceOutput
def run(self, params: GetPriceInput) -> GetPriceOutput:
data = pf_client.trading.get_price(params.symbol)
signal = "bullish" if data['change_24h'] > 2 else \
"bearish" if data['change_24h'] < -2 else "neutral"
return GetPriceOutput(
symbol=params.symbol,
price=data['price'],
change_24h=data['change_24h'],
funding_rate=data['funding_rate'],
signal=signal
)
class OpenTradeTool(BaseTool):
input_schema = OpenTradeInput
output_schema = OpenTradeOutput
def run(self, params: OpenTradeInput) -> OpenTradeOutput:
price = pf_client.trading.get_price(params.symbol)['price']
trade = pf_client.trading.open_trade(
symbol=params.symbol,
side=params.side,
size=params.size_usd / price,
leverage=params.leverage
)
return OpenTradeOutput(
position_id=trade['position_id'],
symbol=params.symbol,
side=params.side,
fill_price=trade['fill_price'],
liquidation_price=trade['liquidation_price'],
estimated_pnl=0
)
Build an Atomic Trading Agent
from atomic_agents.agents.base_agent import BaseAgent, BaseAgentConfig
from atomic_agents.lib.components.system_prompt_generator import SystemPromptGenerator
from instructor import from_openai
from openai import OpenAI
price_tool = GetPriceTool(BaseToolConfig())
trade_tool = OpenTradeTool(BaseToolConfig())
llm_client = from_openai(OpenAI())
prompt_generator = SystemPromptGenerator(
background=[
"You are a disciplined crypto trading agent.",
"You check prices before making trading decisions.",
"You never risk more than 5% of portfolio per trade."
],
steps=[
"Check the price and signal for the requested market.",
"If signal is 'bullish', open a long. If 'bearish', open a short.",
"If 'neutral', do not trade and explain why."
],
output_instructions=[
"Always report: symbol, action taken, and reasoning."
]
)
agent = BaseAgent(
config=BaseAgentConfig(
client=llm_client,
model="gpt-4o",
system_prompt_generator=prompt_generator,
tools=[price_tool, trade_tool]
)
)
response = agent.run("Evaluate BTC and make a trading decision.")
print(response)