Estratégia OSIRIS - IA Autônoma Institucional
Timeframe
5m
Direction
Long Only
Stoploss
-2.0%
Trailing Stop
Yes
ROI
0m: 4.0%, 30m: 2.5%, 60m: 1.5%, 120m: 0.5%
Interface Version
N/A
Startup Candles
200
Indicators
7
freqtrade/freqtrade-strategies
Strategy 003 author@: Gerald Lonlas github@: https://github.com/freqtrade/freqtrade-strategies
"""
OSIRIS - Estratégia Autônoma com IA Institucional
O Freqtrade é apenas o motor de execução.
TODA decisão real vem do OSIRIS Core (Claude API + Knowledge + Learning + Fund Manager).
Fluxo:
1. Freqtrade calcula indicadores técnicos
2. OSIRIS Core recebe os dados + contexto profundo
3. Claude API analisa tudo e decide
4. Risk Manager valida
5. Freqtrade executa
"""
import json
import logging
import os
import sys
from typing import Optional
import numpy as np
from freqtrade.strategy import IStrategy, merge_informative_pair
from freqtrade.strategy import CategoricalParameter, DecimalParameter, IntParameter
from freqtrade.persistence import Trade
from pandas import DataFrame
import talib.abstract as ta
logger = logging.getLogger(__name__)
# Diretório strategies/ para encontrar ai_brain
_strategy_dir = os.path.dirname(os.path.abspath(__file__))
# Import OSIRIS Core (inicializado lazy para evitar problemas no startup)
_osiris_core = None
def get_osiris_core():
"""Inicializa OSIRIS Core sob demanda."""
global _osiris_core
if _osiris_core is None:
try:
# Garantir que strategies/ está no sys.path para importar ai_brain
if _strategy_dir not in sys.path:
sys.path.insert(0, _strategy_dir)
from ai_brain import OsirisCore
# Procurar config em vários locais possíveis
config = {}
config_candidates = [
"osiris_ai_config.json",
os.path.join(os.path.dirname(_strategy_dir), "..", "osiris_ai_config.json"),
os.path.join(os.getcwd(), "osiris_ai_config.json"),
]
for config_path in config_candidates:
if os.path.exists(config_path):
with open(config_path, "r", encoding="utf-8") as f:
config = json.load(f)
break
# API key pode vir do config ou do env
if "anthropic_api_key" not in config:
config["anthropic_api_key"] = os.environ.get("ANTHROPIC_API_KEY", "")
_osiris_core = OsirisCore(config=config)
logger.info("OSIRIS Core inicializado com sucesso")
except Exception as e:
logger.error("Falha ao inicializar OSIRIS Core: %s", e)
_osiris_core = None
return _osiris_core
class OsirisStrategy(IStrategy):
"""
Estratégia OSIRIS - IA Autônoma Institucional
O Freqtrade fornece infraestrutura (dados, execução, backtesting).
O OSIRIS Core fornece inteligência (decisões, knowledge, learning).
Em modo offline (sem API key), usa indicadores técnicos tradicionais como fallback.
"""
# Timeframe principal
timeframe = "5m"
# ROI - Definido pela IA, mas com fallback conservador
minimal_roi = {
"0": 0.04,
"30": 0.025,
"60": 0.015,
"120": 0.005,
}
# Stop Loss - Override pela IA em cada trade
stoploss = -0.02
# Trailing Stop
trailing_stop = True
trailing_stop_positive = 0.01
trailing_stop_positive_offset = 0.015
trailing_only_offset_is_reached = True
# Candles necessários
startup_candle_count = 200
# Parâmetros otimizáveis (fallback mode)
buy_rsi = IntParameter(20, 40, default=30, space="buy", optimize=True)
buy_rsi_max = IntParameter(50, 70, default=65, space="buy", optimize=True)
buy_volume_factor = DecimalParameter(1.0, 3.0, default=1.5, space="buy", optimize=True)
sell_rsi = IntParameter(65, 85, default=75, space="sell", optimize=True)
# Estado da IA
_last_ai_decisions: dict = {} # pair -> AIDecision
def informative_pairs(self):
pairs = self.dp.current_whitelist()
return [(pair, "15m") for pair in pairs]
def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
"""Calcula todos os indicadores técnicos (usados pela IA e como fallback)."""
# RSI
dataframe["rsi"] = ta.RSI(dataframe, timeperiod=14)
# MACD
macd = ta.MACD(dataframe, fastperiod=12, slowperiod=26, signalperiod=9)
dataframe["macd"] = macd["macd"]
dataframe["macd_signal"] = macd["macdsignal"]
dataframe["macd_hist"] = macd["macdhist"]
# Bollinger Bands
bollinger = ta.BBANDS(dataframe, timeperiod=20, nbdevup=2.0, nbdevdn=2.0)
dataframe["bb_upper"] = bollinger["upperband"]
dataframe["bb_middle"] = bollinger["middleband"]
dataframe["bb_lower"] = bollinger["lowerband"]
dataframe["bb_width"] = (dataframe["bb_upper"] - dataframe["bb_lower"]) / dataframe["bb_middle"]
# EMAs
dataframe["ema_8"] = ta.EMA(dataframe, timeperiod=8)
dataframe["ema_21"] = ta.EMA(dataframe, timeperiod=21)
dataframe["ema_50"] = ta.EMA(dataframe, timeperiod=50)
dataframe["ema_200"] = ta.EMA(dataframe, timeperiod=200)
# Volume
dataframe["volume_mean"] = dataframe["volume"].rolling(window=20).mean()
# ATR
dataframe["atr"] = ta.ATR(dataframe, timeperiod=14)
# Stochastic RSI
stoch_rsi = ta.STOCHRSI(dataframe, timeperiod=14, fastk_period=3, fastd_period=3)
dataframe["stoch_rsi_k"] = stoch_rsi["fastk"]
dataframe["stoch_rsi_d"] = stoch_rsi["fastd"]
# Informative 15m
if self.dp:
informative_15m = self.dp.get_pair_dataframe(pair=metadata["pair"], timeframe="15m")
if not informative_15m.empty:
informative_15m["rsi_15m"] = ta.RSI(informative_15m, timeperiod=14)
macd_15m = ta.MACD(informative_15m, fastperiod=12, slowperiod=26, signalperiod=9)
informative_15m["macd_15m"] = macd_15m["macd"]
informative_15m["macd_signal_15m"] = macd_15m["macdsignal"]
informative_15m["ema_50_15m"] = ta.EMA(informative_15m, timeperiod=50)
dataframe = merge_informative_pair(
dataframe, informative_15m, self.timeframe, "15m", ffill=True
)
# === CONSULTAR OSIRIS AI (na última candle apenas) ===
osiris = get_osiris_core()
if osiris and len(dataframe) > 0:
self._run_ai_analysis(dataframe, metadata)
return dataframe
def _run_ai_analysis(self, dataframe: DataFrame, metadata: dict):
"""Consulta o OSIRIS Core para decisão na candle atual."""
pair = metadata["pair"]
osiris = get_osiris_core()
if not osiris:
return
try:
last = dataframe.iloc[-1]
# Montar indicadores
indicators = {
"current_price": float(last["close"]),
"change_24h": float((last["close"] - dataframe.iloc[-288]["close"]) / dataframe.iloc[-288]["close"] * 100) if len(dataframe) >= 288 else 0,
"rsi": float(last.get("rsi", 50)),
"macd": float(last.get("macd", 0)),
"macd_signal": float(last.get("macd_signal", 0)),
"macd_hist": float(last.get("macd_hist", 0)),
"bb_upper": float(last.get("bb_upper", 0)),
"bb_middle": float(last.get("bb_middle", 0)),
"bb_lower": float(last.get("bb_lower", 0)),
"bb_width": float(last.get("bb_width", 0)),
"ema_8": float(last.get("ema_8", 0)),
"ema_21": float(last.get("ema_21", 0)),
"ema_50": float(last.get("ema_50", 0)),
"atr": float(last.get("atr", 0)),
"volume": float(last.get("volume", 0)),
"volume_avg": float(last.get("volume_mean", 0)),
"stoch_rsi_k": float(last.get("stoch_rsi_k", 50)),
"stoch_rsi_d": float(last.get("stoch_rsi_d", 50)),
"volume_ratio": float(last["volume"] / last["volume_mean"]) if last.get("volume_mean", 0) > 0 else 1,
}
# Montar candles (últimos 50)
candle_data = []
for _, row in dataframe.tail(50).iterrows():
candle_data.append({
"open": float(row["open"]),
"high": float(row["high"]),
"low": float(row["low"]),
"close": float(row["close"]),
"volume": float(row["volume"]),
})
# Posição atual
position = None
trades = Trade.get_trades_proxy(pair=pair, is_open=True)
if trades:
t = trades[0]
position = {
"is_open": True,
"open_rate": t.open_rate,
"profit_ratio": t.calc_profit_ratio(last["close"]),
"profit_abs": t.calc_profit(last["close"]),
"stop_loss_abs": t.stop_loss,
"trade_duration_minutes": int((dataframe.iloc[-1].name - t.open_date).total_seconds() / 60) if hasattr(t, 'open_date') else 0,
}
# DECISÃO DO OSIRIS
decision = osiris.decide(
pair=pair,
timeframe=self.timeframe,
indicators=indicators,
candles=candle_data,
position=position,
)
self._last_ai_decisions[pair] = decision
# Salvar nota de mercado (se a IA gerou uma)
if hasattr(decision, 'raw') and decision.raw.get("market_note"):
osiris.knowledge.add_market_note(pair, decision.raw["market_note"])
except Exception as e:
logger.error("OSIRIS AI analysis failed for %s: %s", pair, e)
def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
"""
Entrada: IA decide se deve comprar.
Fallback: indicadores tradicionais se IA não disponível.
"""
pair = metadata["pair"]
decision = self._last_ai_decisions.get(pair)
if decision and decision.should_enter:
# IA diz para comprar — marcar última candle
dataframe.loc[dataframe.index[-1], "enter_long"] = 1
logger.info("OSIRIS AI → ENTER LONG %s (conf=%.0f%%, strategy=%s)",
pair, decision.confidence * 100, decision.strategy_detected)
elif not decision or not get_osiris_core():
# Fallback: indicadores tradicionais
conditions = []
conditions.append(
(dataframe["rsi"] > self.buy_rsi.value) &
(dataframe["rsi"] < self.buy_rsi_max.value)
)
conditions.append(
(dataframe["macd_hist"] > 0) |
(dataframe["macd_hist"] > dataframe["macd_hist"].shift(1))
)
conditions.append(dataframe["close"] > dataframe["bb_middle"])
conditions.append(dataframe["volume"] > dataframe["volume_mean"] * self.buy_volume_factor.value)
conditions.append(dataframe["ema_8"] > dataframe["ema_21"])
conditions.append(dataframe["volume"] > 0)
if conditions:
dataframe.loc[np.all(conditions, axis=0), "enter_long"] = 1
return dataframe
def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
"""
Saída: IA decide se deve vender/fechar.
Fallback: indicadores tradicionais.
"""
pair = metadata["pair"]
decision = self._last_ai_decisions.get(pair)
if decision and decision.should_exit:
dataframe.loc[dataframe.index[-1], "exit_long"] = 1
logger.info("OSIRIS AI → EXIT LONG %s (reason=%s)", pair, decision.reasoning[:80])
elif not decision or not get_osiris_core():
# Fallback
conditions = []
conditions.append(dataframe["rsi"] > self.sell_rsi.value)
conditions.append(
(dataframe["macd_hist"] < 0) &
(dataframe["macd_hist"] < dataframe["macd_hist"].shift(1))
)
if conditions:
dataframe.loc[np.any(conditions, axis=0), "exit_long"] = 1
return dataframe
def confirm_trade_entry(self, pair, order_type, amount, rate, time_in_force, current_time, entry_tag, side, **kwargs) -> bool:
"""Última confirmação antes de abrir trade — OSIRIS valida."""
osiris = get_osiris_core()
decision = self._last_ai_decisions.get(pair)
if osiris and decision and decision.should_enter:
# IA aprovou — notificar OSIRIS que o trade entrou
osiris.on_trade_enter(pair, amount * rate, rate, {"current_price": rate})
return True
if not osiris:
return True # Fallback mode — sem IA, aceitar
if not decision:
# IA carregada mas sem decisão para este par (análise falhou)
# Permitir trade de fallback indicators
return True
return False # IA retornou decisão mas não aprovou
def confirm_trade_exit(self, pair, trade, order_type, amount, rate, time_in_force, exit_reason, current_time, **kwargs) -> bool:
"""Confirmação antes de fechar trade."""
osiris = get_osiris_core()
if osiris:
# Notificar OSIRIS
entry_indicators = {"current_price": trade.open_rate}
exit_indicators = {"current_price": rate}
ai_decision = self._last_ai_decisions.get(pair)
osiris.on_trade_exit(
pair=pair,
amount=amount * rate,
profit=trade.calc_profit(rate),
profit_pct=trade.calc_profit_ratio(rate) * 100,
entry_indicators=entry_indicators,
exit_indicators=exit_indicators,
ai_decision=ai_decision.raw if ai_decision else None,
exit_reason=exit_reason,
)
return True
def custom_stoploss(self, pair: str, trade: Trade, current_time, current_rate, current_profit, after_fill, **kwargs) -> Optional[float]:
"""Stop loss definido pela IA (se disponível)."""
decision = self._last_ai_decisions.get(pair)
if decision and decision.stop_loss and trade.open_rate:
# Converter preço absoluto para % relativo
sl_pct = (decision.stop_loss - trade.open_rate) / trade.open_rate
return sl_pct
return None # Usar stoploss padrão