Test code

[abolfazlbakhti69-art]

‏import requests
‏import hmac
‏import hashlib
‏import time
‏import numpy as np
‏import os
‏import json
‏import threading
‏from datetime import datetime
‏import concurrent.futures

===== تنظیمات اصلی =====

‏ACCESS_ID = os.environ.get("COINEX_API_KEY")
‏SECRET_KEY = os.environ.get("COINEX_SECRET_KEY")
‏BASE_URL = "https://api.coinex.com"

===== لیست ارزها و پارامترهای اختصاصی =====

‏SYMBOL_CONFIG = {
‏ "SOLUSDT": {
‏ "profit_factor": 0.008,
‏ "risk_ceiling": 0.15,
‏ "volatility_threshold": 0.02,
‏ "min_hold_time": 300 # 5 دقیقه
},
‏ "AVAXUSDT": {
‏ "profit_factor": 0.009,
‏ "risk_ceiling": 0.18,
‏ "volatility_threshold": 0.025,
‏ "min_hold_time": 240 # 4 دقیقه
},
‏ "MATICUSDT": {
‏ "profit_factor": 0.007,
‏ "risk_ceiling": 0.12,
‏ "volatility_threshold": 0.018,
‏ "min_hold_time": 360 # 6 دقیقه
},
‏ "DOTUSDT": {
‏ "profit_factor": 0.0065,
‏ "risk_ceiling": 0.13,
‏ "volatility_threshold": 0.015,
‏ "min_hold_time": 420 # 7 دقیقه
},
‏ "ADAUSDT": {
‏ "profit_factor": 0.0085,
‏ "risk_ceiling": 0.16,
‏ "volatility_threshold": 0.022,
‏ "min_hold_time": 300 # 5 دقیقه
}
}

===== مدیریت سرمایه =====

‏INITIAL_TOTAL_BALANCE = 60.0 # سرمایه کل
‏PROFIT_TARGET = 0.35 # هدف سود 35%
‏MAX_DAILY_LOSS = 0.15 # توقف ضرر 15%
‏RISK_FLOOR = 0.04 # حداقل ریسک

===== سیستم لاگ‌گیری حرفه‌ای =====

‏class Logger:
‏ def init(self):
‏ self.log_file = f"multi_asset_log_{datetime.now().strftime('%Y%m%d_%H%M')}.txt"
‏ self.start_time = time.time()
‏ self.lock = threading.Lock()

‏ def log(self, message, symbol="GLOBAL", level="INFO"):
‏ with self.lock:
‏ timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
‏ log_entry = f"[{timestamp}] [{symbol}] [{level}] {message}"
‏ print(log_entry)
‏ with open(self.log_file, "a") as f:
‏ f.write(log_entry + "\n")

‏ def performance_report(self, final_balance, symbol_reports):
‏ duration = time.time() - self.start_time
‏ profit = final_balance - INITIAL_TOTAL_BALANCE

‏ report = f"\n{'='60}\n"
‏ report += f"🚀 MULTI-ASSET PERFORMANCE REPORT | Runtime: {duration/3600:.2f} hrs\n"
‏ report += f"💵 Initial Total Balance: ${INITIAL_TOTAL_BALANCE:.2f}\n"
‏ report += f"💰 Final Total Balance: ${final_balance:.2f}\n"
‏ report += f"📈 Total Profit: ${profit:.2f} ({profit/INITIAL_TOTAL_BALANCE100:.2f}%)\n"
‏ report += f"{'-'*60}\n"

‏ for symbol, data in symbol_reports.items():
‏ report += (
‏ f"🔹 {symbol}: "
‏ f"Start: ${data['start_balance']:.2f} | "
‏ f"End: ${data['end_balance']:.2f} | "
‏ f"Profit: {data['profit_percent']:.2f}% | "
‏ f"Trades: {data['trades']} | "
‏ f"Win Rate: {data['win_rate']:.1f}%\n"
)

‏ report += f"{'='*60}"
‏ self.log(report)

===== موتور معاملاتی برای هر ارز =====

‏class AssetTrader:
‏ def init(self, symbol, initial_balance, config, logger):
‏ self.symbol = symbol
‏ self.balance = initial_balance
‏ self.config = config
‏ self.logger = logger
‏ self.trade_history = []
‏ self.consecutive_wins = 0
‏ self.consecutive_losses = 0
‏ self.start_balance = initial_balance
‏ self.market_conditions = {
‏ "volatility": 0.01,
‏ "trend_strength": 0,
‏ "volume_ratio": 1.0
}

‏ def get_historical_prices(self, minutes=30):
"""دریافت داده‌های تاریخی برای ارز خاص"""
‏ url = f"{BASE_URL}/v1/market/kline"
‏ params = {"market": self.symbol, "type": "5min", "limit": min(200, minutes//5)}

‏ try:
‏ resp = requests.get(url, params=params, timeout=3)
‏ data = resp.json()
‏ if data.get("code") == 0:
‏ return [float(entry[4]) for entry in data["data"]]
‏ except:
‏ pass
‏ return []

‏ def get_realtime_data(self):
"""دریافت داده‌های لحظه‌ای برای ارز خاص"""
‏ url = f"{BASE_URL}/v1/market/ticker"
‏ params = {"market": self.symbol}

‏ try:
‏ resp = requests.get(url, params=params, timeout=2)
‏ data = resp.json()
‏ if data.get("code") == 0:
‏ ticker = data["data"]["ticker"]
‏ return {
‏ "price": float(ticker["last"]),
‏ "high": float(ticker["high"]),
‏ "low": float(ticker["low"]),
‏ "volume": float(ticker["vol"])
}
‏ except:
‏ pass
‏ return None

‏ def calculate_technical_indicators(self, prices):
"""محاسبه اندیکاتورهای پیشرفته"""
‏ if len(prices) < 20:
‏ return {"rsi": 50, "macd": 0, "atr": 0.01}

    # محاسبه RSI

‏ deltas = np.diff(prices)
‏ 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:])
‏ rsi = 100 - (100 / (1 + (avg_gain / avg_loss))) if avg_loss != 0 else 100

    # محاسبه MACD

‏ ema12 = np.mean(prices[-12:])
‏ ema26 = np.mean(prices[-26:])
‏ macd = ema12 - ema26

    # محاسبه ATR

‏ tr = [max(prices[i]-prices[i-1], abs(prices[i]-prices[i-1]), abs(prices[i]-prices[i-1]))
‏ for i in range(1, len(prices))]
‏ atr = np.mean(tr[-14:])

‏ return {"rsi": rsi, "macd": macd, "atr": atr}

‏ def update_market_conditions(self):
"""بروزرسانی شرایط بازار برای این ارز"""
‏ prices = self.get_historical_prices(120)
‏ realtime_data = self.get_realtime_data()

‏ if not prices or not realtime_data:
‏ return False

‏ returns = np.diff(prices) / prices[:-1]
‏ self.market_conditions["volatility"] = np.std(returns)

‏ sma_short = np.mean(prices[-6:])
‏ sma_long = np.mean(prices[-24:])
‏ self.market_conditions["trend_strength"] = (sma_short - sma_long) / sma_long

‏ avg_volume = np.mean([prices[i] for i in range(-10, 0)])
‏ self.market_conditions["volume_ratio"] = realtime_data["volume"] / avg_volume

‏ return True

‏ def calculate_dynamic_risk(self):
"""مدیریت ریسک پویا برای این ارز"""
‏ base_risk = 0.08

‏ if abs(self.market_conditions["trend_strength"]) > 0.01:
‏ base_risk = min(self.config["risk_ceiling"], base_risk * 1.5)

‏ if self.market_conditions["volatility"] > self.config["volatility_threshold"]:
‏ base_risk = max(RISK_FLOOR, base_risk * 0.7)

‏ if self.consecutive_wins >= 3:
‏ return min(self.config["risk_ceiling"], base_risk * (1 + self.consecutive_wins * 0.1))
‏ elif self.consecutive_losses >= 2:
‏ return max(RISK_FLOOR, base_risk * (1 - self.consecutive_losses * 0.15))

‏ return base_risk

‏ def calculate_profit_target(self):
"""حد سود پویا برای این ارز"""
‏ base_target = self.config["profit_factor"]

‏ if abs(self.market_conditions["trend_strength"]) > 0.015:
‏ base_target *= 1.4

‏ if self.market_conditions["volume_ratio"] > 1.5:
‏ base_target *= 1.3

‏ if self.market_conditions["volatility"] > self.config["volatility_threshold"]:
‏ base_target *= 0.8

‏ return max(0.006, min(0.015, base_target))

‏ def generate_trade_signal(self, current_price, indicators):
"""تولید سیگنال معاملاتی برای این ارز"""
‏ buy_conditions = (
‏ indicators["rsi"] < 65 and
‏ indicators["macd"] > 0 and
‏ self.market_conditions["trend_strength"] > 0.005 and
‏ self.market_conditions["volume_ratio"] > 1.2
)

‏ sell_conditions = (
‏ indicators["rsi"] > 35 and
‏ indicators["macd"] < 0 and
‏ self.market_conditions["trend_strength"] < -0.005 and
‏ self.market_conditions["volume_ratio"] > 1.2
)

‏ if buy_conditions:
‏ return "BUY"
‏ elif sell_conditions:
‏ return "SELL"
‏ return "NEUTRAL"

‏ def execute_trade(self):
"""اجرای معامله برای این ارز"""
‏ try:
‏ current_data = self.get_realtime_data()
‏ if not current_data:
‏ return False

‏ prices = self.get_historical_prices(30)
‏ if len(prices) < 20:
‏ return False

‏ indicators = self.calculate_technical_indicators(prices)
‏ self.update_market_conditions()

‏ signal = self.generate_trade_signal(current_data["price"], indicators)
‏ if signal == "NEUTRAL":
‏ return False

‏ risk_percent = self.calculate_dynamic_risk()
‏ profit_target_percent = self.calculate_profit_target()
‏ stop_loss_percent = profit_target_percent * 0.7

‏ trade_amount = min(self.balance * risk_percent, self.balance * 0.9)
‏ amount = trade_amount / current_data["price"]

‏ entry_price = current_data["price"]

‏ if signal == "BUY":
‏ profit_target = entry_price * (1 + profit_target_percent)
‏ stop_loss = entry_price * (1 - stop_loss_percent)
‏ else:
‏ profit_target = entry_price * (1 - profit_target_percent)
‏ stop_loss = entry_price * (1 + stop_loss_percent)

‏ trade = {
‏ 'entry': entry_price,
‏ 'amount': amount,
‏ 'signal': signal,
‏ 'profit_target': profit_target,
‏ 'stop_loss': stop_loss,
‏ 'start_time': time.time(),
‏ 'result': None,
‏ 'profit': 0
}

‏ self.monitor_trade(trade)
‏ return True

‏ except Exception as e:
‏ self.logger.log(f"خطا در اجرای معامله: {str(e)}", self.symbol, "ERROR")
‏ return False

‏ def monitor_trade(self, trade):
‏ position_open = True
‏ best_price = trade['entry']
‏ last_update = time.time()

‏ self.logger.log(f"🔹 معامله جدید | {trade['signal']} | حجم: {trade['amount']:.4f} | ورود: {trade['entry']:.4f}", self.symbol)

‏ while position_open:
‏ try:
‏ current_data = self.get_realtime_data()
‏ if not current_data:
‏ time.sleep(1)
‏ continue

‏ current_price = current_data["price"]

‏ if (trade['signal'] == "BUY" and current_price > best_price) or
‏ (trade['signal'] == "SELL" and current_price < best_price):
‏ best_price = current_price

‏ if trade['signal'] == "BUY":
‏ unrealized_profit = (current_price - trade['entry']) * trade['amount']
‏ else:
‏ unrealized_profit = (trade['entry'] - current_price) * trade['amount']

‏ if time.time() - last_update > 10:
‏ self.logger.log(f"📊 قیمت: {current_price:.4f} | سود شناور: {unrealized_profit:.4f} USD", self.symbol)
‏ last_update = time.time()

‏ time_in_trade = time.time() - trade['start_time']

            # شرایط خروج

‏ if (trade['signal'] == "BUY" and current_price >= trade['profit_target']) or
‏ (trade['signal'] == "SELL" and current_price <= trade['profit_target']):
‏ trade['result'] = "WIN"
‏ trade['profit'] = unrealized_profit
‏ position_open = False
‏ self.logger.log(f"🎯 هدف سود فعال شد! سود: {unrealized_profit:.4f} USD", self.symbol)

‏ elif (trade['signal'] == "BUY" and current_price <= trade['stop_loss']) or
‏ (trade['signal'] == "SELL" and current_price >= trade['stop_loss']):
‏ trade['result'] = "LOSS"
‏ trade['profit'] = unrealized_profit
‏ position_open = False
‏ self.logger.log(f"⛔ حد ضرر فعال شد! ضرر: {abs(unrealized_profit):.4f} USD", self.symbol)

‏ elif time_in_trade > 30:
‏ dynamic_factor = 0.6 + (min(time_in_trade, 300) / 600) * 0.4

‏ if trade['signal'] == "BUY":
‏ dynamic_stop = best_price * (1 - (trade['stop_loss'] * dynamic_factor))
‏ if current_price <= dynamic_stop:
‏ trade['result'] = "WIN"
‏ trade['profit'] = unrealized_profit
‏ position_open = False
‏ self.logger.log(f"🔒 تریلینگ استاپ فعال شد! سود: {unrealized_profit:.4f} USD", self.symbol)
‏ else:
‏ dynamic_stop = best_price * (1 + (trade['stop_loss'] * dynamic_factor))
‏ if current_price >= dynamic_stop:
‏ trade['result'] = "WIN"
‏ trade['profit'] = unrealized_profit
‏ position_open = False
‏ self.logger.log(f"🔒 تریلینگ استاپ فعال شد! سود: {unrealized_profit:.4f} USD", self.symbol)

‏ elif time_in_trade > self.config["min_hold_time"]:
‏ trade['result'] = "WIN" if unrealized_profit > 0 else "LOSS"
‏ trade['profit'] = unrealized_profit
‏ position_open = False
‏ self.logger.log(f"⏱️ خروج زمانی! {'سود' if unrealized_profit > 0 else 'ضرر'}: {abs(unrealized_profit):.4f} USD", self.symbol)

‏ time.sleep(0.5)

‏ except Exception as e:
‏ self.logger.log(f"خطا در نظارت: {str(e)}", self.symbol, "ERROR")
‏ time.sleep(2)

‏ self.balance += trade['profit']
‏ self.trade_history.append(trade)

‏ if trade['result'] == "WIN":
‏ self.consecutive_wins += 1
‏ self.consecutive_losses = 0
‏ else:
‏ self.consecutive_losses += 1
‏ self.consecutive_wins = 0

‏ self.logger.log(f"موجودی: ${self.balance:.2f} | برد: {self.consecutive_wins} | باخت: {self.consecutive_losses}", self.symbol)

‏ def run(self, runtime_hours=10):
‏ start_time = time.time()
‏ last_trade_time = 0

‏ self.logger.log(f"🚀 آغاز معاملات | سرمایه: ${self.balance:.2f}", self.symbol)

‏ while (time.time() < start_time + (runtime_hours * 3600) and
‏ self.balance > self.start_balance * (1 - MAX_DAILY_LOSS) and
‏ self.balance < self.start_balance * (1 + PROFIT_TARGET)):

‏ if time.time() - last_trade_time < 60:
‏ time.sleep(1)
‏ continue

‏ self.update_market_conditions()
‏ trade_executed = self.execute_trade()
‏ last_trade_time = time.time()

‏ if not trade_executed:
‏ sleep_time = 30 if self.market_conditions["volatility"] > self.config["volatility_threshold"] else 15
‏ time.sleep(sleep_time)

‏ return {
‏ "start_balance": self.start_balance,
‏ "end_balance": self.balance,
‏ "profit_percent": (self.balance - self.start_balance) / self.start_balance * 100,
‏ "trades": len(self.trade_history),
‏ "win_rate": len([t for t in self.trade_history if t['result'] == "WIN"]) / len(self.trade_history) * 100 if self.trade_history else 0
}

===== سیستم مدیریت چندارزی =====

‏class MultiAssetManager:
‏ def init(self, total_balance, symbol_config, runtime_hours=10):
‏ self.logger = Logger()
‏ self.runtime_hours = runtime_hours
‏ self.symbol_config = symbol_config
‏ self.total_balance = total_balance
‏ self.symbol_balances = self.initialize_balances()

‏ def initialize_balances(self):
"""تقسیم سرمایه به صورت مساوی بین ارزها"""
‏ num_symbols = len(self.symbol_config)
‏ balance_per_symbol = self.total_balance / num_symbols
‏ return {symbol: balance_per_symbol for symbol in self.symbol_config.keys()}

‏ def run_single_asset(self, symbol):
"""اجرای تریدر برای یک ارز خاص"""
‏ trader = AssetTrader(
‏ symbol=symbol,
‏ initial_balance=self.symbol_balances[symbol],
‏ config=self.symbol_config[symbol],
‏ logger=self.logger
)
‏ return symbol, trader.run(self.runtime_hours)

‏ def execute(self):
"""اجرای همزمان معاملات برای تمام ارزها"""
‏ start_time = time.time()
‏ self.logger.log(f"🚀 آغاز سیستم معاملاتی چندارزی | سرمایه کل: ${self.total_balance:.2f} | مدت: {self.runtime_hours} ساعت")

    # اجرای موازی معاملات برای هر ارز

‏ with concurrent.futures.ThreadPoolExecutor(max_workers=len(self.symbol_config)) as executor:
‏ futures = {executor.submit(self.run_single_asset, symbol): symbol for symbol in self.symbol_config}
‏ results = {}

‏ for future in concurrent.futures.as_completed(futures):
‏ symbol = futures[future]
‏ try:
‏ results[symbol] = future.result()[1]
‏ except Exception as e:
‏ self.logger.log(f"خطا در اجرای {symbol}: {str(e)}", "SYSTEM", "ERROR")
‏ results[symbol] = {
‏ "start_balance": self.symbol_balances[symbol],
‏ "end_balance": self.symbol_balances[symbol],
‏ "profit_percent": 0,
‏ "trades": 0,
‏ "win_rate": 0
}

    # محاسبه مجموع نتایج

‏ final_total_balance = sum(data["end_balance"] for data in results.values())
‏ runtime = time.time() - start_time

    # گزارش نهایی

‏ self.logger.performance_report(final_total_balance, results)

‏ return {
‏ "total_runtime": runtime,
‏ "final_balance": final_total_balance,
‏ "symbol_results": results
}

===== اجرای برنامه =====

‏if name == "main":
# محاسبه سرمایه هر ارز (60 دلار کل / 5 ارز = 12 دلار برای هر ارز)
‏ manager = MultiAssetManager(
‏ total_balance=INITIAL_TOTAL_BALANCE,
‏ symbol_config=SYMBOL_CONFIG,
‏ runtime_hours=10
)

‏ results = manager.execute()
‏ print(f"\n✅ اجرا با موفقیت به پایان رسید | سرمایه نهایی: ${results['final_balance']:.2f}")