دليل تطوير نظام التداول الآلي للبيتكوين

1. تهيئة بيئة التطوير

<br/>

# تثبيت المكتبات المطلوبة
pip install ccxt pandas numpy talib TA-Lib python-dotenv schedule websockets flask

• الأدوات الرئيسية:
  • CCXT: واجهة برمجة تطبيقات موحدة لأكثر من 100 بورصة
  • Pandas: تحليل البيانات
  • TA-Lib: حساب المؤشرات الفنية
  • Websockets: استقبال البيانات في الوقت الفعلي
  • Flask: بناء لوحة التحكم

2. اختيار البورصة وإعداد واجهة برمجة التطبيقات (API)

<br/>

# حفظ مفتاح API في ملف .env
API_KEY = "your_api_key"  # مفتاح API الصادر من البورصة
API_SECRET = "your_api_secret"  # سر API الصادر من البورصة

• البورصات الموصى بها:
  • Binance: سيولة عالية، دعم REST/Websocket
  • Bybit: متخصصة في تداول العقود الآجلة
  • Upbit: استقرار الكيان القانوني المحلي

3. جمع بيانات الرسوم البيانية

import ccxt

<br/>

# تهيئة واجهة برمجة تطبيقات Binance
binance = ccxt.binance({
    'apiKey': API_KEY,  # تعيين مفتاح API
    'secret': API_SECRET,  # تعيين سر API
    'enableRateLimit': True  # تمكين حد طلبات API
})

<br/>

# استقبال بيانات OHLCV في الوقت الفعلي (Websocket)
async def fetch_btc_data():
    async with websockets.connect('wss://fstream.binance.com/ws/btcusdt@kline_1m') as ws:
        while True:
            data = await ws.recv()  # استقبال البيانات في الوقت الفعلي
            print(json.loads(data))  # طباعة البيانات

4. تطوير استراتيجية التداول

4-1. تطبيق المؤشرات الفنية

import talib

<br/>

# دالة لحساب المؤشرات الفنية
def calculate_indicators(df):
    # حساب المتوسط المتحرك البسيط (SMA) لمدة 20 يومًا
    df['MA20'] = talib.SMA(df['close'], timeperiod=20)
    # حساب مؤشر القوة النسبية (RSI) لمدة 14 يومًا
    df['RSI'] = talib.RSI(df['close'], timeperiod=14)
    # حساب MACD (تباعد تقارب المتوسطات المتحركة)
    df['MACD'], _, _ = talib.MACD(df['close'])
    return df

4-2. منطق إنشاء إشارات التداول

<br/>

# دالة لإنشاء إشارات التداول
def generate_signal(df):
    latest = df.iloc[-1]  # الحصول على أحدث البيانات
    
    # استراتيجية المتوسط المتحرك المزدوج (Dual SMA)
    if latest['MA20'] > latest['MA50'] and df['MA20'].iloc[-2] <= df['MA50'].iloc[-2]:
        return 'BUY'  # إشارة شراء
    elif latest['MA20'] < latest['MA50'] and df['MA20'].iloc[-2] >= df['MA50'].iloc[-2]:
        return 'SELL'  # إشارة بيع
    else:
        return 'HOLD'  # إشارة احتفاظ

5. نظام إدارة المخاطر

<br/>

# دالة لحساب حجم الصفقة
def calculate_position_size(balance, risk_per_trade=0.02):
    # حساب حجم الصفقة بناءً على رأس المال الإجمالي ونسبة المخاطرة لكل صفقة
    return balance * risk_per_trade

<br/>

# دالة لتعيين وقف الخسارة
def set_stop_loss(entry_price, atr, multiplier=1.5):
    # حساب وقف الخسارة بناءً على ATR (متوسط المدى الحقيقي)
    return entry_price - (atr * multiplier)

6. وحدة تنفيذ الأوامر

<br/>

# دالة لتنفيذ أمر
def execute_order(side, amount, symbol='BTC/USDT'):
    try:
        if side == 'BUY':
            # تنفيذ أمر شراء سوقي
            order = binance.create_market_buy_order(symbol, amount)
        elif side == 'SELL':
            # تنفيذ أمر بيع سوقي
            order = binance.create_market_sell_order(symbol, amount)
        print(f"Order Executed: {order}")  # طباعة سجل تنفيذ الأمر
        return order
    except Exception as e:
        print(f"Order Failed: {e}")  # طباعة رسالة خطأ إذا فشل الأمر
        return None

7. نظام الاختبار الخلفي (Backtesting)

<br/>

# دالة الاختبار الخلفي
def backtest_strategy(df, initial_balance=10000):
    balance = initial_balance  # تعيين رأس المال الأولي
    position = 0  # تهيئة المركز
    
    for i in range(1, len(df)):
        signal = df['signal'].iloc[i]  # الحصول على إشارة التداول
        price = df['close'].iloc[i]  # الحصول على سعر الإغلاق
        
        if signal == 'BUY' and position == 0:
            # تعيين المركز عند تلقي إشارة شراء
            position = balance / price
            balance = 0
        elif signal == 'SELL' and position > 0:
            # تصفية المركز عند تلقي إشارة بيع
            balance = position * price
            position = 0
            
    return balance  # إرجاع الرصيد النهائي

8. نظام المراقبة والتسجيل (Logging)

import logging

<br/>

# إعدادات التسجيل
logging.basicConfig(filename='trading.log', level=logging.INFO)

<br/>

# دالة تسجيل المعاملات
def log_transaction(order):
    logging.info(f"""
    [تفاصيل المعاملة]
    الوقت: {datetime.now()}
    النوع: {order['side']}
    المبلغ: {order['amount']}
    السعر: {order['price']}
    الحالة: {order['status']}
    """)

9. الدالة الرئيسية

<br/>

# حلقة التنفيذ الرئيسية
def main():
    while True:
        try:
            df = fetch_realtime_data()  # جلب البيانات في الوقت الفعلي
            df = calculate_indicators(df)  # حساب المؤشرات الفنية
            signal = generate_signal(df)  # إنشاء إشارة التداول
            
            if signal != 'HOLD':
                # تنفيذ أمر إذا كانت هناك إشارة تداول
                amount = calculate_position_size(get_balance())
                execute_order(signal, amount)
                
            time.sleep(60)  # التشغيل على فترات دقيقة واحدة
        except KeyboardInterrupt:
            break  # الخروج من الحلقة عند مقاطعة المستخدم

10. تدابير تعزيز الأمان

10-1. إدارة مفاتيح API

• لا تقم أبدًا بتثبيتها في نظام إدارة الإصدارات
• استخدم AWS Secrets Manager أو Hashicorp Vault.

10-2. تشفير حركة المرور

binance = ccxt.binance({
    'options': {'adjustForTimeDifference': True},  # تصحيح فرق التوقيت
    'proxies': {'https': 'http://10.10.1.10:3128'}  # إعدادات الوكيل
})

11. مثال على النشر

<br/>

# تسجيل خدمة النظام (Linux)
[Unit]
Description=بوت تداول العملات المشفرة
After=network.target

[Service]
ExecStart=/usr/bin/python3 /path/to/bot.py
Restart=always

[Install]
WantedBy=multi-user.target

📌 ملاحظات هامة

• تقييد المخاطر بأقل من 1٪ من رأس المال الأولي
• الامتثال لحد معدل واجهة برمجة تطبيقات البورصة (Rate Limit) بدقة
• الاستعداد لتقلبات السوق في عطلات نهاية الأسبوع / العطلات الرسمية
• إعادة تقييم أداء الاستراتيجية بشكل دوري
• يجب إجراء اختبار في بيئة افتراضية لمدة أسبوعين على الأقل قبل استثمار الأموال الفعلية
• يجب استلام عنوان API الفعلي وتطبيقه لتشغيله