Tháng 3/2026, một nhà giao dịch crypto độc lập tên Minh (giấu tên) đứng trước màn hình với danh mục 5 altcoin và vô số chart. Anh ấy dành 6 tiếng mỗi ngày để phân tích thủ công — nhưng vẫn bỏ lỡ tín hiệu vàng vì thông tin đến quá chậm. Sau 3 tuần xây dựng hệ thống LangChain + Crypto API + AI, anh ấy tự động hóa hoàn toàn quy trình phân tích với độ trễ dưới 2 giây. Bài viết này sẽ hướng dẫn bạn xây dựng hệ thống tương tự.
Tại Sao Kết Hợp LangChain và Crypto API?
LangChain là framework mạnh mẽ để xây dựng ứng dụng dựa trên LLM, nhưng để phân tích crypto thời gian thực, bạn cần:
- Kết nối API thị trường: Giá, khối lượng, orderbook, funding rate
- Xử lý dữ liệu có cấu trúc: Chuyển đổi JSON thành ngữ cảnh cho LLM
- Tạo tín hiệu giao dịch: BUY/SELL/HOLD với confidence score
- Theo dõi danh mục: P&L, allocation, risk metrics
Kiến Trúc Hệ Thống Hoàn Chỉnh
Hệ thống gồm 4 module chính hoạt động theo pipeline:
┌─────────────────────────────────────────────────────────────────┐
│ CRYPTO AI SIGNAL SYSTEM │
├─────────────────────────────────────────────────────────────────┤
│ [Data Collector] → [Preprocessor] → [LLM Analyzer] → [Signaler]│
│ ↓ ↓ ↓ ↓ │
│ CryptoAPI LangChain HolySheep AI Alert │
│ (Binance/ Chains & (DeepSeek V3.2) System │
│ CoinGecko) Prompting ↓ │
│ [Telegram/Discord] │
└─────────────────────────────────────────────────────────────────┘Cài Đặt Môi Trường
# Tạo virtual environment
python -m venv crypto-ai-env
source crypto-ai-env/bin/activate # Linux/Mac
crypto-ai-env\Scripts\activate # Windows
Cài đặt dependencies
pip install langchain langchain-community
pip install langchain-huggingface
pip install requests aiohttp websockets
pip install python-dotenv pandas numpy
pip install ta-lib # Technical Analysis (cài đặt phức tạp hơn)
Package thay thế nếu TA-Lib gặp lỗi cài đặt
pip install ta fintaModule 1: Kết Nối API Crypto - Data Collector
Chúng ta sử dụng Binance API cho futures và CoinGecko API cho giá spot. Dưới đây là implementation hoàn chỉnh:
"""
Crypto Data Collector - Kết nối Binance & CoinGecko
Sử dụng HolySheep AI cho phân tích
"""
import requests
import asyncio
import aiohttp
from typing import Dict, List, Optional
from datetime import datetime
import json
class CryptoDataCollector:
"""Thu thập dữ liệu từ nhiều nguồn API crypto"""
def __init__(self, holy_sheep_api_key: str):
self.holy_sheep_api_key = holy_sheep_api_key
self.base_url_holysheep = "https://api.holysheep.ai/v1"
self.binance_spot = "https://api.binance.com/api/v3"
self.binance_futures = "https://fapi.binance.com/fapi/v1"
self.coingecko_base = "https://api.coingecko.com/api/v3"
async def get_binance_klines(self, symbol: str, interval: str = "1h", limit: int = 100) -> List[Dict]:
"""Lấy dữ liệu nến từ Binance Spot"""
url = f"{self.binance_spot}/klines"
params = {
"symbol": symbol.upper(),
"interval": interval,
"limit": limit
}
async with aiohttp.ClientSession() as session:
async with session.get(url, params=params) as response:
if response.status == 200:
data = await response.json()
return self._parse_klines(data, symbol)
else:
raise Exception(f"Binance API Error: {response.status}")
def _parse_klines(self, data: List, symbol: str) -> List[Dict]:
"""Parse dữ liệu kline từ Binance"""
parsed = []
for k in data:
parsed.append({
"symbol": symbol,
"open_time": datetime.fromtimestamp(k[0] / 1000),
"open": float(k[1]),
"high": float(k[2]),
"low": float(k[3]),
"close": float(k[4]),
"volume": float(k[5]),
"close_time": datetime.fromtimestamp(k[6] / 1000),
"quote_volume": float(k[7])
})
return parsed
async def get_funding_rate(self, symbol: str) -> Optional[Dict]:
"""Lấy funding rate từ Binance Futures - chỉ báo sentiment quan trọng"""
url = f"{self.binance_futures}/premiumIndex"
async with aiohttp.ClientSession() as session:
async with session.get(url) as response:
if response.status == 200:
data = await response.json()
for item in data:
if item['symbol'] == symbol.upper():
return {
"symbol": symbol,
"funding_rate": float(item['lastFundingRate']) * 100,
"next_funding_time": item['nextFundingTime'],
"mark_price": float(item['markPrice']),
"index_price": float(item['indexPrice'])
}
return None
async def get_orderbook(self, symbol: str, limit: int = 20) -> Dict:
"""Lấy orderbook depth"""
url = f"{self.binance_spot}/depth"
params = {"symbol": symbol.upper(), "limit": limit}
async with aiohttp.ClientSession() as session:
async with session.get(url, params=params) as response:
if response.status == 200:
data = await response.json()
return {
"symbol": symbol,
"bids": [[float(p), float(q)] for p, q in data['bids']],
"asks": [[float(p), float(q)] for p, q in data['asks']],
"bid_depth": sum(float(q) for _, q in data['bids'][:5]),
"ask_depth": sum(float(q) for _, q in data['asks'][:5])
}
async def get_coingecko_market_data(self, coin_ids: List[str]) -> List[Dict]:
"""Lấy dữ liệu thị trường từ CoinGecko - miễn phí, không rate limit quá cao"""
url = f"{self.coingecko_base}/coins/markets"
params = {
"vs_currency": "usd",
"ids": ",".join(coin_ids),
"order": "market_cap_desc",
"sparkline": "true",
"price_change_percentage": "1h,24h,7d"
}
async with aiohttp.ClientSession() as session:
async with session.get(url, params=params) as response:
if response.status == 200:
return await response.json()
else:
return []
==================== SỬ DỤNG HOLYSHEEP AI ====================
async def analyze_with_holysheep(api_key: str, prompt: str) -> str:
"""Gửi prompt đến HolySheep AI cho phân tích"""
headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
payload = {
"model": "deepseek-v3.2", # Model rẻ nhất, phù hợp cho data analysis
"messages": [
{"role": "system", "content": "Bạn là chuyên gia phân tích crypto. Phân tích dữ liệu và đưa ra tín hiệu giao dịch rõ ràng."},
{"role": "user", "content": prompt}
],
"temperature": 0.3, # Low temperature cho phân tích nhất quán
"max_tokens": 500
}
async with aiohttp.ClientSession() as session:
async with session.post(
"https://api.holysheep.ai/v1/chat/completions",
headers=headers,
json=payload
) as response:
result = await response.json()
return result['choices'][0]['message']['content']
==================== VÍ DỤ SỬ DỤNG ====================
async def main():
collector = CryptoDataCollector("YOUR_HOLYSHEEP_API_KEY")
# Thu thập dữ liệu BTC/USDT
klines = await collector.get_binance_klines("BTCUSDT", "1h", 100)
funding = await collector.get_funding_rate("BTCUSDT")
orderbook = await collector.get_orderbook("BTCUSDT", 20)
market = await collector.get_coingecko_market_data(["bitcoin", "ethereum"])
print(f"Đã thu thập {len(klines)} nến")
print(f"Funding Rate: {funding['funding_rate'] if funding else 'N/A'}%")
print(f"Market Cap Rank: BTC #{market[0]['market_cap_rank']}")
if __name__ == "__main__":
asyncio.run(main())Module 2: LangChain Integration - Data Preprocessor
Đây là phần core của hệ thống - sử dụng LangChain Agents để xử lý dữ liệu và tạo prompts thông minh:
"""
LangChain Crypto Analyzer - Phân tích dữ liệu với LLM
"""
import os
from langchain.agents import AgentType, initialize_agent, Tool
from langchain.agents.agent_toolkits import ZapierToolkit
from langchain.tools import StructuredTool
from langchain.callbacks import get_openai_callback
from langchain.prompts import PromptTemplate
from langchain.chat_models import ChatOpenAI
from langchain.schema import HumanMessage, SystemMessage
from pydantic import BaseModel, Field
from typing import List, Dict, Optional
import json
import asyncio
==================== HOLYSHEEP API CONFIGURATION ====================
⚠️ QUAN TRỌNG: Sử dụng HolySheep thay vì OpenAI
Đăng ký tại: https://www.holysheep.ai/register
class HolySheepLLM:
"""Wrapper cho HolySheep AI API - tương thích LangChain"""
def __init__(self, api_key: str):
self.api_key = api_key
self.base_url = "https://api.holysheep.ai/v1"
def __call__(self, messages: List) -> str:
"""Implement LangChain callback interface"""
import aiohttp
# Convert LangChain messages format to OpenAI format
openai_messages = []
for msg in messages:
if isinstance(msg, SystemMessage):
openai_messages.append({"role": "system", "content": msg.content})
elif isinstance(msg, HumanMessage):
openai_messages.append({"role": "user", "content": msg.content})
else:
openai_messages.append({"role": "assistant", "content": msg.content})
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
payload = {
"model": "deepseek-v3.2", # Model rẻ nhất, hiệu quả cho phân tích
"messages": openai_messages,
"temperature": 0.2,
"max_tokens": 800
}
# Synchronous call for LangChain compatibility
import requests
response = requests.post(
f"{self.base_url}/chat/completions",
headers=headers,
json=payload,
timeout=30
)
if response.status_code == 200:
return response.json()['choices'][0]['message']['content']
else:
raise Exception(f"HolySheep API Error: {response.status_code} - {response.text}")
class TechnicalIndicators(BaseModel):
"""Kết quả phân tích kỹ thuật"""
symbol: str = Field(description="Mã crypto")
rsi: float = Field(description="RSI - Relative Strength Index (0-100)")
macd: Dict[str, float] = Field(description="MACD values: macd, signal, histogram")
sma_20: float = Field(description="SMA 20 ngày")
sma_50: float = Field(description="SMA 50 ngày")
bb_upper: float = Field(description="Bollinger Upper Band")
bb_lower: float = Field(description="Bollinger Lower Band")
volume_ratio: float = Field(description="Tỷ lệ volume so với trung bình")
def calculate_indicators(klines: List[Dict]) -> TechnicalIndicators:
"""Tính toán các chỉ báo kỹ thuật từ dữ liệu nến"""
import numpy as np
closes = np.array([float(k['close']) for k in klines])
volumes = np.array([float(k['volume']) for k in klines])
# RSI Calculation
deltas = np.diff(closes)
gains = np.where(deltas > 0, deltas, 0)
losses = np.where(deltas < 0, -deltas, 0)
avg_gain = np.mean(gains[-14:])
avg_loss = np.mean(losses[-14:])
rs = avg_gain / (avg_loss + 1e-10)
rsi = 100 - (100 / (1 + rs))
# SMA
sma_20 = np.mean(closes[-20:]) if len(closes) >= 20 else closes[-1]
sma_50 = np.mean(closes[-50:]) if len(closes) >= 50 else sma_20
# Bollinger Bands
std = np.std(closes[-20:])
bb_upper = sma_20 + (2 * std)
bb_lower = sma_20 - (2 * std)
# MACD (simplified)
ema_12 = np.mean(closes[-12:]) # Simplified EMA
ema_26 = np.mean(closes[-26:]) if len(closes) >= 26 else ema_12
macd_line = ema_12 - ema_26
signal_line = macd_line * 0.9 # Simplified signal
histogram = macd_line - signal_line
# Volume ratio
avg_volume = np.mean(volumes[-20:])
volume_ratio = volumes[-1] / avg_volume if avg_volume > 0 else 1
return TechnicalIndicators(
symbol=klines[-1]['symbol'],
rsi=float(rsi),
macd={"macd": float(macd_line), "signal": float(signal_line), "histogram": float(histogram)},
sma_20=float(sma_20),
sma_50=float(sma_50),
bb_upper=float(bb_upper),
bb_lower=float(bb_lower),
volume_ratio=float(volume_ratio)
)
class CryptoSignalGenerator:
"""Generator tín hiệu giao dịch sử dụng LangChain + HolySheep AI"""
def __init__(self, holysheep_api_key: str):
self.llm = HolySheepLLM(holysheep_api_key)
self.signal_prompt = PromptTemplate.from_template("""
Bạn là chuyên gia phân tích crypto với 10 năm kinh nghiệm giao dịch.
DỮ LIỆU PHÂN TÍCH:
Symbol: {symbol}
Giá hiện tại: ${price}
RSI (14): {rsi:.2f}
MACD: {macd}
SMA 20: ${sma_20:.2f}
SMA 50: ${sma_50:.2f}
Bollinger Bands: ${bb_lower:.2f} - ${bb_upper:.2f}
Volume Ratio: {volume_ratio:.2f}
Funding Rate: {funding_rate}%
Market Cap: ${market_cap}
Dominance: {dominance}%
YÊU CẦU:
Phân tích dữ liệu trên và đưa ra:
1. **Tín hiệu**: BUY / SELL / HOLD (với confidence %)
2. **Entry Price**: Giá vào lệnh đề xuất
3. **Stop Loss**: Mức cắt lỗ
4. **Take Profit**: Mức chốt lời
5. **Risk/Reward Ratio**: Tỷ lệ R:R
6. **Thời gian giữ lệnh**: recommendation (short-term/medium-term/long-term)
7. **Lý do**: Giải thích ngắn gọn logic đằng sau tín hiệu
Format response JSON như sau:
{{
"signal": "BUY/SELL/HOLD",
"confidence": 0-100,
"entry_price": number,
"stop_loss": number,
"take_profit": number,
"risk_reward": number,
"holding_period": "short/medium/long",
"reasoning": "text"
}}
""")
def generate_signal(self, market_data: Dict, indicators: TechnicalIndicators,
funding_rate: float, orderbook: Dict) -> Dict:
"""Tạo tín hiệu giao dịch"""
# Format MACD data
macd_str = f"MACD: {indicators.macd['macd']:.2f}, Signal: {indicators.macd['signal']:.2f}, Histogram: {indicators.macd['histogram']:.2f}"
prompt = self.signal_prompt.format(
symbol=market_data.get('symbol', 'BTC'),
price=market_data.get('current_price', 0),
rsi=indicators.rsi,
macd=macd_str,
sma_20=indicators.sma_20,
sma_50=indicators.sma_50,
bb_upper=indicators.bb_upper,
bb_lower=indicators.bb_lower,
volume_ratio=indicators.volume_ratio,
funding_rate=funding_rate,
market_cap=market_data.get('market_cap', 0),
dominance=market_data.get('btc_dominance', 0)
)
response = self.llm([
SystemMessage(content="Bạn là chuyên gia phân tích crypto. Chỉ trả lời JSON."),
HumanMessage(content=prompt)
])
# Parse JSON response
try:
signal_data = json.loads(response)
return {
**signal_data,
"raw_indicators": indicators.dict(),
"orderbook_sentiment": "BULLISH" if orderbook.get('bid_depth', 0) > orderbook.get('ask_depth', 0) else "BEARISH"
}
except json.JSONDecodeError:
return {
"signal": "HOLD",
"confidence": 50,
"error": "Failed to parse LLM response",
"raw_response": response
}
==================== VÍ DỤ SỬ DỤNG ====================
def main():
# Khởi tạo với HolySheep API Key
HOLYSHEEP_API_KEY = os.getenv("HOLYSHEEP_API_KEY", "YOUR_HOLYSHEEP_API_KEY")
generator = CryptoSignalGenerator(HOLYSHEEP_API_KEY)
# Dữ liệu mẫu (thay bằng dữ liệu thực từ collector)
sample_market_data = {
"symbol": "BTC",
"current_price": 67500.00,
"market_cap": 1320000000000,
"btc_dominance": 52.5
}
# Giả lập indicators (thay bằng calculate_indicators thực tế)
sample_indicators = TechnicalIndicators(
symbol="BTC",
rsi=68.5,
macd={"macd": 245.30, "signal": 198.45, "histogram": 46.85},
sma_20=66800.00,
sma_50=65200.00,
bb_upper=68500.00,
bb_lower=65100.00,
volume_ratio=1.35
)
signal = generator.generate_signal(
market_data=sample_market_data,
indicators=sample_indicators,
funding_rate=0.012,
orderbook={"bid_depth": 1500000, "ask_depth": 1200000}
)
print("=" * 50)
print(f"SYMBOL: {sample_market_data['symbol']}")
print(f"SIGNAL: {signal['signal']} (Confidence: {signal.get('confidence', 0)}%)")
print(f"Entry: ${signal.get('entry_price', 0)}")
print(f"Stop Loss: ${signal.get('stop_loss', 0)}")
print(f"Take Profit: ${signal.get('take_profit', 0)}")
print(f"R/R Ratio: {signal.get('risk_reward', 0)}")
print("=" * 50)
if __name__ == "__main__":
main()Module 3: Signal Alert System - Telegram/Discord Integration
Hệ thống gửi tín hiệu qua Telegram với format chuyên nghiệp:
"""
Crypto Signal Alert System - Gửi thông báo qua Telegram/Discord
"""
import asyncio
import aiohttp
import time
from datetime import datetime
from typing import Dict, List, Optional
from enum import Enum
class SignalStrength(Enum):
STRONG_BUY = "🟢🟢🟢 STRONG BUY"
BUY = "🟢 BUY"
NEUTRAL = "🟡 NEUTRAL"
SELL = "🔴 SELL"
STRONG_SELL = "🔴🔴🔴 STRONG SELL"
class AlertFormatter:
"""Format tín hiệu thành message đẹp"""
@staticmethod
def format_telegram_message(signal_data: Dict) -> str:
"""Format signal thành Telegram message với emoji"""
signal = signal_data.get('signal', 'HOLD')
confidence = signal_data.get('confidence', 0)
# Xác định strength
if signal == "BUY":
if confidence >= 75:
strength = SignalStrength.STRONG_BUY
else:
strength = SignalStrength.BUY
elif signal == "SELL":
if confidence >= 75:
strength = SignalStrength.STRONG_SELL
else:
strength = SignalStrength.SELL
else:
strength = SignalStrength.NEUTRAL
entry = signal_data.get('entry_price', 0)
sl = signal_data.get('stop_loss', 0)
tp = signal_data.get('take_profit', 0)
rr = signal_data.get('risk_reward', 0)
message = f"""
📊 *CRYPTO SIGNAL ALERT*
━━━━━━━━━━━━━━━━━━━━━━
🔹 *Symbol:* {signal_data.get('symbol', 'N/A')}
🔹 *Signal:* {strength.value}
🔹 *Confidence:* {confidence}%
📈 *ENTRY & EXIT*
• Entry: ${entry:,.2f}
• Stop Loss: ${sl:,.2f} (-{((entry-sl)/entry*100):.1f}%)
• Take Profit: ${tp:,.2f} (+{((tp-entry)/entry*100):.1f}%)
📐 *RISK MANAGEMENT*
• Risk/Reward: 1:{rr:.2f}
• Position Size: Max 2% equity
• Holding: {signal_data.get('holding_period', 'medium')}
💡 *REASONING:*
_{signal_data.get('reasoning', 'No analysis available')}_
━━━━━━━━━━━━━━━━━━━━━━
🕐 Generated: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}
🤖 Powered by HolySheep AI
"""
return message.strip()
@staticmethod
def format_discord_embed(signal_data: Dict) -> Dict:
"""Format signal thành Discord Embed"""
signal = signal_data.get('signal', 'HOLD')
confidence = signal_data.get('confidence', 0)
# Color based on signal
color_map = {
'STRONG_BUY': 0x00FF00,
'BUY': 0x32CD32,
'NEUTRAL': 0xFFD700,
'SELL': 0xFF4500,
'STRONG_SELL': 0xFF0000
}
color = 0xFFD700 # Default yellow
if signal == "BUY":
color = 0x00FF00 if confidence >= 75 else 0x32CD32
elif signal == "SELL":
color = 0xFF0000 if confidence >= 75 else 0xFF4500
entry = signal_data.get('entry_price', 0)
sl = signal_data.get('stop_loss', 0)
tp = signal_data.get('take_profit', 0)
embed = {
"title": f"📊 {signal_data.get('symbol', 'N/A')} - {signal} Signal",
"color": color,
"fields": [
{"name": "Confidence", "value": f"{confidence}%", "inline": True},
{"name": "Holding Period", "value": signal_data.get('holding_period', 'medium'), "inline": True},
{"name": "Entry Price", "value": f"${entry:,.2f}", "inline": True},
{"name": "Stop Loss", "value": f"${sl:,.2f}", "inline": True},
{"name": "Take Profit", "value": f"${tp:,.2f}", "inline": True},
{"name": "Risk/Reward", "value": f"1:{signal_data.get('risk_reward', 0):.2f}", "inline": True},
{"name": "Reasoning", "value": signal_data.get('reasoning', 'N/A'), "inline": False}
],
"footer": {
"text": f"Generated by HolySheep AI | {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}"
}
}
return embed
class AlertSender:
"""Gửi alerts qua nhiều kênh"""
def __init__(self, telegram_token: str = None, telegram_chat_id: str = None,
discord_webhook: str = None):
self.telegram_token = telegram_token
self.telegram_chat_id = telegram_chat_id
self.discord_webhook = discord_webhook
async def send_telegram(self, message: str) -> bool:
"""Gửi message qua Telegram Bot API"""
if not self.telegram_token or not self.telegram_chat_id:
print("Telegram not configured, skipping...")
return False
url = f"https://api.telegram.org/bot{self.telegram_token}/sendMessage"
payload = {
"chat_id": self.telegram_chat_id,
"text": message,
"parse_mode": "Markdown"
}
try:
async with aiohttp.ClientSession() as session:
async with session.post(url, json=payload) as response:
if response.status == 200:
print(f"✅ Telegram message sent successfully")
return True
else:
print(f"❌ Telegram error: {response.status}")
return False
except Exception as e:
print(f"❌ Telegram exception: {e}")
return False
async def send_discord(self, embed: Dict) -> bool:
"""Gửi embed qua Discord Webhook"""
if not self.discord_webhook:
print("Discord not configured, skipping...")
return False
try:
async with aiohttp.ClientSession() as session:
async with session.post(
self.discord_webhook,
json={"embeds": [embed]}
) as response:
if response.status == 204 or response.status == 200:
print(f"✅ Discord message sent successfully")
return True
else:
print(f"❌ Discord error: {response.status}")
return False
except Exception as e:
print(f"❌ Discord exception: {e}")
return False
async def send_alert(self, signal_data: Dict) -> Dict:
"""Gửi alert đến tất cả các kênh đã cấu hình"""
results = {}
# Format messages
telegram_msg = AlertFormatter.format_telegram_message(signal_data)
discord_embed = AlertFormatter.format_discord_embed(signal_data)
# Send to Telegram
if self.telegram_token:
results['telegram'] = await self.send_telegram(telegram_msg)
# Send to Discord
if self.discord_webhook:
results['discord'] = await self.send_discord(discord_embed)
return results
==================== VÍ DỤ SỬ DỤNG ====================
async def main():
# Khởi tạo alert sender
sender = AlertSender(
telegram_token="YOUR_TELEGRAM_BOT_TOKEN",
telegram_chat_id="YOUR_CHAT_ID",
discord_webhook="YOUR_DISCORD_WEBHOOK_URL"
)
# Sample signal data
sample_signal = {
"symbol": "BTCUSDT",
"signal": "BUY",
"confidence": 82,
"entry_price": 67500.00,
"stop_loss": 66200.00,
"take_profit": 70200.00,
"risk_reward": 2.08,
"holding_period": "medium",
"reasoning": "RSI đang ở mức 68.5 (trung lập-bullish), MACD histogram dương, giá đang trên SMA 20 và SMA 50. Funding rate thấp cho thấy ít người short. Volume tăng 35% so với trung bình."
}
# Gửi alert
results = await sender.send_alert(sample_signal)
print(f"Alert sent: {results}")
if __name__ == "__main__":
asyncio.run(main())Module 4: Main Orchestrator - Pipeline Hoàn Chỉnh
Kết hợp tất cả modules và chạy pipeline tự động:
"""
Crypto Signal Bot - Main Orchestrator
Chạy pipeline hoàn chỉnh: Thu thập → Phân tích → Gửi Alert
"""
import os
import asyncio
import aiohttp
from datetime import datetime, timedelta
from typing import List, Dict
from dotenv import load_dotenv
import time
Import các module đã tạo
from crypto_data_collector import CryptoDataCollector
from langchain_analyzer import CryptoSignalGenerator, calculate_indicators, TechnicalIndicators
from alert_system import AlertSender
Load environment variables
load_dotenv()
class CryptoSignalBot:
"""Bot tự động phân tích và gửi tín hiệu crypto"""
def __init__(self):
# API Keys
self.holysheep_api_key = os.getenv("HOLYSHEEP_API_KEY")
self.telegram_token = os.getenv("TELEGRAM_BOT_TOKEN")
self.telegram_chat_id = os.getenv("TELEGRAM_CHAT_ID")
self.discord_webhook = os.getenv("DISCORD_WEBHOOK_URL")
# Initialize components
self.collector = CryptoDataCollector(self.holysheep_api_key)
self.generator = CryptoSignalGenerator(self.holysheep_api_key)
self.alert_sender = AlertSender(
telegram_token=self.telegram_token,
telegram_chat_id=self.telegram_chat_id,
discord_webhook=self.discord_webhook
)
# Watchlist
self.watchlist = [
{"symbol": "BTCUSDT", "coin_id": "bitcoin", "weight": 0.3},
{"