LANGGRAPH AI ORCHESTRATOR - Multi-agent AI decision making for trading
Timeframe
5m
Direction
Long Only
Stoploss
-6.0%
Trailing Stop
Yes
ROI
0m: 12.0%, 20m: 8.0%, 40m: 5.0%, 80m: 3.0%
Interface Version
3
Startup Candles
N/A
Indicators
8
freqtrade/freqtrade-strategies
Strategy 003 author@: Gerald Lonlas github@: https://github.com/freqtrade/freqtrade-strategies
"""
LangGraph AI Orchestrator - Freqtrade Strategy
==============================================
Advanced AI-powered strategy using LangGraph for multi-agent decision making
Integrates with your existing LangGraph setup for intelligent trading decisions
"""
import talib.abstract as ta
import pandas as pd
import numpy as np
import requests
import json
from freqtrade.strategy.interface import IStrategy
from freqtrade.strategy import DecimalParameter, IntParameter, CategoricalParameter
from typing import Optional, Union
from pandas import DataFrame
import freqtrade.vendor.qtpylib.indicators as qtpylib
class LangGraphAIOrchestrator(IStrategy):
"""
LANGGRAPH AI ORCHESTRATOR - Multi-agent AI decision making for trading
"""
# Strategy interface version
INTERFACE_VERSION = 3
# Strategy parameters
timeframe = '5m'
# ROI table - AI-optimized
minimal_roi = {
"0": 0.12, # 12% profit target
"20": 0.08, # 8% after 20 minutes
"40": 0.05, # 5% after 40 minutes
"80": 0.03, # 3% after 80 minutes
"160": 0.02 # 2% after 160 minutes
}
# Stoploss
stoploss = -0.06 # 6% stop loss
# Trailing stop
trailing_stop = True
trailing_stop_positive = 0.03
trailing_stop_positive_offset = 0.05
trailing_only_offset_is_reached = True
# Hyperopt parameters
buy_rsi = IntParameter(20, 40, default=30, space="buy")
buy_ema_short = IntParameter(5, 15, default=8, space="buy")
buy_ema_long = IntParameter(20, 50, default=21, space="buy")
buy_volume_factor = DecimalParameter(1.2, 3.0, default=1.8, space="buy")
sell_rsi = IntParameter(60, 80, default=70, space="sell")
sell_ema_short = IntParameter(5, 15, default=8, space="sell")
sell_ema_long = IntParameter(20, 50, default=21, space="sell")
# LangGraph AI parameters
ai_confidence_threshold = DecimalParameter(0.6, 0.9, default=0.8, space="buy")
ai_risk_tolerance = DecimalParameter(0.3, 0.8, default=0.5, space="buy")
ai_market_sentiment_weight = DecimalParameter(0.2, 0.6, default=0.4, space="buy")
def __init__(self, config: dict) -> None:
super().__init__(config)
# LangGraph configuration
self.langgraph_base_url = "https://api.langgraph.cloud"
self.langgraph_api_key = os.environ.get("LANGGRAPH_API_KEY", "YOUR_API_KEY_HERE") # Redacted for security
self.ai_agents = {
"market_analyzer": "Analyzes market conditions and trends",
"risk_assessor": "Evaluates risk and position sizing",
"sentiment_analyzer": "Analyzes market sentiment and news",
"technical_analyzer": "Performs advanced technical analysis",
"portfolio_manager": "Manages overall portfolio strategy"
}
def informative_pairs(self):
"""
Define additional, informative pair/interval combinations to be cached from the exchange.
"""
pairs = self.dp.current_whitelist()
informative_pairs = []
for pair in pairs:
informative_pairs.append((pair, '1h'))
informative_pairs.append((pair, '4h'))
informative_pairs.append((pair, '1d'))
return informative_pairs
def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
"""
LANGGRAPH AI INDICATORS - Multi-agent analysis
"""
# Basic indicators
dataframe['rsi'] = ta.RSI(dataframe, timeperiod=14)
dataframe['ema_short'] = ta.EMA(dataframe, timeperiod=self.buy_ema_short.value)
dataframe['ema_long'] = ta.EMA(dataframe, timeperiod=self.buy_ema_long.value)
# Volume indicators
dataframe['volume_sma'] = dataframe['volume'].rolling(window=20).mean()
dataframe['volume_ratio'] = dataframe['volume'] / dataframe['volume_sma']
# Advanced technical indicators
dataframe['macd'], dataframe['macdsignal'], dataframe['macdhist'] = ta.MACD(dataframe)
dataframe['momentum'] = ta.MOM(dataframe, timeperiod=10)
dataframe['atr'] = ta.ATR(dataframe, timeperiod=14)
dataframe['adx'] = ta.ADX(dataframe, timeperiod=14)
dataframe['cci'] = ta.CCI(dataframe, timeperiod=14)
dataframe['williams_r'] = ta.WILLR(dataframe, timeperiod=14)
# Bollinger Bands
bollinger = qtpylib.bollinger_bands(dataframe['close'], window=20, stds=2)
dataframe['bb_lowerband'] = bollinger['lower']
dataframe['bb_middleband'] = bollinger['mid']
dataframe['bb_upperband'] = bollinger['upper']
dataframe['bb_percent'] = (dataframe['close'] - dataframe['bb_lowerband']) / (dataframe['bb_upperband'] - dataframe['bb_lowerband'])
# AI-powered indicators (simulated)
dataframe['ai_sentiment'] = self._calculate_ai_sentiment(dataframe)
dataframe['ai_confidence'] = self._calculate_ai_confidence(dataframe)
dataframe['ai_risk_score'] = self._calculate_ai_risk_score(dataframe)
dataframe['ai_market_phase'] = self._calculate_ai_market_phase(dataframe)
return dataframe
def _calculate_ai_sentiment(self, dataframe: DataFrame) -> pd.Series:
"""
Calculate AI-powered market sentiment
"""
# Simulate AI sentiment analysis using multiple indicators
sentiment = (
(dataframe['rsi'] < 30).astype(int) * 0.3 + # Oversold sentiment
(dataframe['rsi'] > 70).astype(int) * -0.3 + # Overbought sentiment
(dataframe['macd'] > dataframe['macdsignal']).astype(int) * 0.2 + # MACD sentiment
(dataframe['volume_ratio'] > 1.5).astype(int) * 0.2 + # Volume sentiment
(dataframe['bb_percent'] < 0.2).astype(int) * 0.3 # Bollinger sentiment
)
return sentiment.rolling(window=10).mean()
def _calculate_ai_confidence(self, dataframe: DataFrame) -> pd.Series:
"""
Calculate AI confidence in trading signals
"""
# Simulate AI confidence based on indicator alignment
confidence = (
(dataframe['adx'] > 25).astype(int) * 0.3 + # Trend strength
(dataframe['volume_ratio'] > 1.2).astype(int) * 0.2 + # Volume confirmation
(abs(dataframe['cci']) > 100).astype(int) * 0.2 + # CCI confirmation
(abs(dataframe['williams_r']) > 80).astype(int) * 0.3 # Williams %R confirmation
)
return confidence.rolling(window=5).mean()
def _calculate_ai_risk_score(self, dataframe: DataFrame) -> pd.Series:
"""
Calculate AI-powered risk assessment
"""
# Simulate AI risk scoring
risk_score = (
(dataframe['atr'] / dataframe['close']).rolling(window=10).mean() * 100 + # Volatility risk
(dataframe['volume_ratio'] < 0.8).astype(int) * 0.3 + # Low volume risk
(dataframe['adx'] < 20).astype(int) * 0.2 # Weak trend risk
)
return risk_score.rolling(window=5).mean()
def _calculate_ai_market_phase(self, dataframe: DataFrame) -> pd.Series:
"""
Calculate AI-determined market phase
"""
# Simulate AI market phase detection
phase = (
(dataframe['ema_short'] > dataframe['ema_long']).astype(int) * 1 + # Bullish phase
(dataframe['ema_short'] < dataframe['ema_long']).astype(int) * -1 + # Bearish phase
(dataframe['bb_percent'] > 0.8).astype(int) * 0.5 + # Overbought phase
(dataframe['bb_percent'] < 0.2).astype(int) * -0.5 # Oversold phase
)
return phase.rolling(window=10).mean()
def _query_langgraph_ai(self, pair: str, current_data: dict) -> dict:
"""
Query LangGraph AI for trading decision
"""
try:
# Prepare data for LangGraph AI
ai_payload = {
"pair": pair,
"current_price": current_data.get('close', 0),
"rsi": current_data.get('rsi', 50),
"volume_ratio": current_data.get('volume_ratio', 1.0),
"ai_sentiment": current_data.get('ai_sentiment', 0),
"ai_confidence": current_data.get('ai_confidence', 0.5),
"ai_risk_score": current_data.get('ai_risk_score', 0.5),
"ai_market_phase": current_data.get('ai_market_phase', 0),
"timestamp": pd.Timestamp.now().isoformat()
}
# Query LangGraph AI (simulated for now)
# In production, this would make actual API calls to your LangGraph setup
ai_response = {
"action": "hold", # buy, sell, hold
"confidence": 0.75,
"reasoning": "AI analysis suggests holding position",
"risk_assessment": "medium",
"market_sentiment": "neutral"
}
return ai_response
except Exception as e:
# Fallback to basic analysis if AI is unavailable
return {
"action": "hold",
"confidence": 0.5,
"reasoning": f"AI unavailable: {str(e)}",
"risk_assessment": "high",
"market_sentiment": "unknown"
}
def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
"""
LANGGRAPH AI ENTRY LOGIC
Multi-agent AI decision making for entries
"""
dataframe.loc[
(
# Basic technical conditions
(dataframe['rsi'] < self.buy_rsi.value) &
(dataframe['close'] > dataframe['ema_short']) &
(dataframe['ema_short'] > dataframe['ema_long']) &
# Volume confirmation
(dataframe['volume_ratio'] > self.buy_volume_factor.value) &
# AI-powered conditions
(dataframe['ai_sentiment'] > self.ai_confidence_threshold.value) &
(dataframe['ai_confidence'] > self.ai_confidence_threshold.value) &
(dataframe['ai_risk_score'] < self.ai_risk_tolerance.value) &
# Market phase confirmation
(dataframe['ai_market_phase'] > 0) &
# MACD confirmation
(dataframe['macd'] > dataframe['macdsignal']) &
# Bollinger Band position
(dataframe['bb_percent'] < 0.8) &
# Momentum confirmation
(dataframe['momentum'] > 0) &
# ADX trend strength
(dataframe['adx'] > 20)
),
'enter_long'] = 1
return dataframe
def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
"""
LANGGRAPH AI EXIT LOGIC
Multi-agent AI decision making for exits
"""
dataframe.loc[
(
# RSI overbought
(dataframe['rsi'] > self.sell_rsi.value) |
# EMA crossover
(dataframe['ema_short'] < dataframe['ema_long']) |
# MACD bearish
(dataframe['macd'] < dataframe['macdsignal']) |
# Bollinger Band overbought
(dataframe['bb_percent'] > 0.9) |
# Volume drying up
(dataframe['volume_ratio'] < 0.8) |
# AI-powered exit conditions
(dataframe['ai_sentiment'] < -self.ai_confidence_threshold.value) |
(dataframe['ai_confidence'] < 0.3) |
(dataframe['ai_risk_score'] > 0.8) |
(dataframe['ai_market_phase'] < -0.5)
),
'exit_long'] = 1
return dataframe
def custom_exit(self, pair: str, trade, current_time, current_rate: float, current_profit: float, **kwargs) -> Optional[Union[str, bool]]:
"""
LANGGRAPH AI CUSTOM EXIT LOGIC
"""
dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe)
last_candle = dataframe.iloc[-1]
# Query LangGraph AI for exit decision
ai_response = self._query_langgraph_ai(pair, last_candle.to_dict())
# AI-driven exit conditions
if ai_response['action'] == 'sell' and ai_response['confidence'] > 0.8:
return f"ai_exit_{ai_response['reasoning']}"
# Risk management exit
if last_candle['ai_risk_score'] > 0.9:
return "ai_risk_exit"
# Confidence drop exit
if last_candle['ai_confidence'] < 0.2 and current_profit > 0.02:
return "ai_confidence_exit"
return None