Spaces:
Sleeping
Sleeping
File size: 4,399 Bytes
8731e23 f2ed1f6 8731e23 f2ed1f6 8731e23 f2ed1f6 8731e23 f2ed1f6 8731e23 f2ed1f6 8731e23 f2ed1f6 8731e23 f2ed1f6 8731e23 f2ed1f6 8731e23 f2ed1f6 8731e23 f2ed1f6 8731e23 f2ed1f6 8731e23 f2ed1f6 8731e23 f2ed1f6 8731e23 f2ed1f6 8731e23 f2ed1f6 8731e23 f2ed1f6 |
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 132 133 134 135 |
import json
import openai
import requests
from pydantic import ValidationError
from models.market_data import MarketData
from models.suggestion import (
TradeSuggestion,
TradeDirection,
TakeProfitPoints,
RecommendationType,
)
class AIService:
@staticmethod
def _prepare_prompt(
symbol, leverage, trade_amount, market_data: MarketData, current_price: float
):
klines = [
{
"timestamp": k.timestamp,
"open": k.open,
"high": k.high,
"low": k.low,
"close": k.close,
"volume": k.volume,
}
for k in market_data.klines
]
return f"""
You are an expert crypto futures trader. Based on market data, suggest a trade for {symbol}.
Current Price: ${current_price}
Leverage: {leverage}x
Amount: ${trade_amount}
Market Data:
{json.dumps(klines, indent=2)}
Only return valid JSON:
{{
"direction": "long" or "short",
"entry_price": float,
"recommended_leverage": int (10-75),
"take_profit": {{
"first": float,
"second": float,
"third": float
}},
"recommendation": "It is recommended to enter the transaction." or "It is not recommended to enter into a transaction." or "Entering the trade with caution"
}}
"""
def _parse_response(
self, response: str, symbol: str, current_price: float, trade_amount: float
):
try:
if response.startswith("```json"):
response = response[7:]
if response.endswith("```"):
response = response[:-3]
data = json.loads(response)
rec_map = {
"It is recommended to enter the transaction.": RecommendationType.RECOMMENDED,
"It is not recommended to enter into a transaction.": RecommendationType.NOT_RECOMMENDED,
"Entering the trade with caution": RecommendationType.CAUTIOUS,
}
return TradeSuggestion(
symbol=symbol,
direction=TradeDirection(data["direction"]),
entry_price=float(data["entry_price"]),
recommended_leverage=int(data["recommended_leverage"]),
take_profit=TakeProfitPoints(**data["take_profit"]),
recommendation=rec_map.get(
data["recommendation"], RecommendationType.CAUTIOUS
),
current_price=current_price,
trade_amount=trade_amount,
)
except Exception as e:
raise ValueError(f"Failed to parse AI response: {e}\nRaw: {response}")
def generate(
self,
symbol,
leverage,
trade_amount,
market_data,
current_price,
provider,
openai_key,
hf_token,
):
prompt = self._prepare_prompt(
symbol, leverage, trade_amount, market_data, current_price
)
if provider == "OpenAI":
openai.api_key = openai_key
try:
res = openai.ChatCompletion.create(
model="gpt-3.5-turbo",
messages=[
{
"role": "system",
"content": "You are a crypto trading expert.",
},
{"role": "user", "content": prompt},
],
temperature=0.2,
)
content = res.choices[0].message.content
return self._parse_response(
content, symbol, current_price, trade_amount
)
except Exception as e:
print(f"OpenAI Error: {e}")
return None
elif provider == "HuggingFace":
try:
headers = {"Authorization": f"Bearer {hf_token}"}
data = {"inputs": prompt}
url = "https://api-inference.huggingface.co/models/tiiuae/falcon-7b"
res = requests.post(url, headers=headers, json=data)
response = res.json()
text = response[0]["generated_text"]
return self._parse_response(text, symbol, current_price, trade_amount)
except Exception as e:
print(f"HuggingFace Error: {e}")
return None
|