abolfazlbakhti69-art opened a new issue, #6495:
URL: https://github.com/apache/jmeter/issues/6495
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_BALANCE*100:.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}")
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
