import gradio as gr
import requests
import torch
import re
from transformers import AutoModelForCausalLM, AutoTokenizer, pipeline
MARKET_CONTEXT = "Market data is loading..."
MODEL_ID = "Qwen/Qwen2.5-1.5B-Instruct"
device = "cuda" if torch.cuda.is_available() else "cpu"
tokenizer = AutoTokenizer.from_pretrained(MODEL_ID)
model = AutoModelForCausalLM.from_pretrained(
MODEL_ID,
torch_dtype="auto",
device_map="auto"
)
def fetch_crypto_data():
url = "https://api.coingecko.com/api/v3/coins/markets"
params = {
"vs_currency": "usd",
"ids": "bitcoin,ethereum,solana,binancecoin",
"order": "market_cap_desc",
"per_page": 4,
"page": 1,
"sparkline": "true",
"price_change_percentage": "24h"
}
global MARKET_CONTEXT
try:
response = requests.get(url, params=params, timeout=10)
if response.status_code != 200:
return None
data = response.json()
context_parts = []
processed_data = []
for coin in data:
symbol = coin['symbol'].upper()
price = coin['current_price']
chg_24 = coin.get('price_change_percentage_24h', 0) or 0
mcap = coin['market_cap'] or 0
history = coin.get('sparkline_in_7d', {}).get('price', [])
context_parts.append(f"[{symbol}: ${price}, 24h:{chg_24:.1f}%]")
processed_data.append({
"name": coin['name'], "symbol": symbol, "price": price,
"chg_24": chg_24, "mcap": mcap, "history": history
})
MARKET_CONTEXT = " | ".join(context_parts)
return processed_data
except Exception:
return None
def chat_logic(user_input, history):
fetch_crypto_data()
system_prompt = f"You are a professional Crypto Assistant. LIVE DATA: {MARKET_CONTEXT}. Answer concisely."
messages = [{"role": "system", "content": system_prompt}]
for human, assistant in history:
messages.append({"role": "user", "content": human})
messages.append({"role": "assistant", "content": assistant})
messages.append({"role": "user", "content": user_input})
text = tokenizer.apply_chat_template(
messages,
tokenize=False,
add_generation_prompt=True
)
model_inputs = tokenizer([text], return_tensors="pt").to(model.device)
generated_ids = model.generate(
**model_inputs,
max_new_tokens=512,
do_sample=True,
temperature=0.7
)
response_ids = [
output_ids[len(input_ids):] for input_ids, output_ids in zip(model_inputs.input_ids, generated_ids)
]
response = tokenizer.batch_decode(response_ids, skip_special_tokens=True)[0]
cleaned_response = re.sub(r'