指定
"""
Fast Multi-Pair RSI Trading Bot
Supports:
- BTCUSDT
- XAUUSD
- GBPUSD
Opens fast buy or sell trades based on RSI signals
Closes trades after 5, 10, or 15 minutes
"""
import asyncio
import time
from dataclasses import dataclass, field
from typing import Dict, List, Optional
import pandas as pd
import numpy as np
# ===== RSI calculation ===== #
def compute_rsi(close: pd.Series, period: int = 14) -> pd.Series:
delta = close.diff()
gain = delta.clip(lower=0)
loss = -delta.clip(upper=0)
avg_gain = gain.ewm(alpha=1 / period, adjust=False).mean()
avg_loss = loss.ewm(alpha=1 / period, adjust=False).mean()
rs = avg_gain / avg_loss
return 100 - (100 / (1 + rs))
# ===== Position structure ===== #
@dataclass
class Position:
id: str
symbol: str
side: str
entry_price: float
size: float
opened_at: float
duration_min: int
# ===== Config ===== #
@dataclass
class BotConfig:
symbols: List[str] = field(default_factory=lambda: ["BTCUSDT", "XAUUSD", "GBPUSD"])
rsi_period: int = 14
rsi_oversold: int = 30
rsi_overbought: int = 70
durations_min: List[int] = field(default_factory=lambda: [5, 10, 15])
account_equity: float = 2000.0
risk_pct: float = 0.5
lot_size: Optional[float] = None
paper: bool = True
# ===== Trading Bot ===== #
class MultiPairRSIBot:
def __init__(self, cfg: BotConfig):
self.cfg = cfg
self.data: Dict[str, pd.DataFrame] = {sym: pd.DataFrame() for sym in cfg.symbols}
self.positions: Dict[str, Dict[str, Position]] = {sym: {} for sym in cfg.symbols}
self._id = 0
# ========== Fake 1-minute feed for PAPER mode ========== #
def get_fake_ohlcv(self, symbol):
now = int(time.time()) * 1000
df = self.data[symbol]
last_close = df["close"].iloc[-1] if not df.empty else 1000 + np.random.rand() * 10
change = np.random.normal(0, 0.0008)
close = last_close * (1 + change)
high = max(last_close, close)
low = min(last_close, close)
return (now, last_close, high, low, close, 0)
# ========== Append new candle ========== #
def append_candle(self, symbol, ohlc):
ts, o, h, l, c, v = ohlc
row = {"timestamp": pd.to_datetime(ts, unit="ms"),
"open": o, "high": h, "low": l, "close": c, "volume": v}
self.data[symbol] = pd.concat([self.data[symbol], pd.DataFrame([row])], ignore_index=True)
if len(self.data[symbol]) > 2000:
self.data[symbol] = self.data[symbol].iloc[-2000:]
# ========== Timeframe aggregation ========== #
def to_tf(self, symbol, minutes):
df = self.data[symbol]
if df.empty:
return pd.DataFrame()
df["bucket"] = df["timestamp"].dt.floor(f"{minutes}T")
out = df.groupby("bucket").agg({
"open": "first",
"high": "max",
"low": "min",
"close": "last",
"volume": "sum"
}).reset_index().rename(columns={"bucket": "timestamp"})
return out
# ========== Position sizing ========== #
def get_size(self, price):
if self.cfg.lot_size:
return self.cfg.lot_size
risk_amount = self.cfg.account_equity * (self.cfg.risk_pct / 100)
return round(risk_amount / price, 4)
# ========== Check RSI signals and enter trades ========== #
async def process_signals(self, symbol):
for dur in self.cfg.durations_min:
df = self.to_tf(symbol, dur)
if len(df) < self.cfg.rsi_period + 2:
continue
df["rsi"] = compute_rsi(df["close"], self.cfg.rsi_period)
prev = df["rsi"].iloc[-2]
last = df["rsi"].iloc[-1]
price = df["close"].iloc[-1]
# BUY: RSI cross up
if prev <= self.cfg.rsi_oversold and last > prev:
size = self.get_size(price)
await self.open_position(symbol, "buy", price, size, dur)
# SELL: RSI cross down
if prev >= self.cfg.rsi_overbought and last < prev:
size = self.get_size(price)
await self.open_position(symbol, "sell", price, size, dur)
# ========== Open position ========== #
async def open_position(self, symbol, side, price, size, duration):
self._id += 1
pid = f"{symbol}_{self._id}"
print(f"[{symbol}] OPEN {side.upper()} @ {price:.2f} | {duration}m | size {size}")
pos = Position(
id=pid,
symbol=symbol,
side=side,
entry_price=price,
size=size,
opened_at=time.time(),
duration_min=duration
)
self.positions[symbol][pid] = pos
# ========== Close expired trades ========== #
async def close_expired(self, symbol):
now = time.time()
to_close = []
for pid, pos in self.positions[symbol].items():
if (now - pos.opened_at) / 60 >= pos.duration_min:
to_close.append(pid)
for pid in to_close:
await self.close_position(symbol, pid)
# ========== Close position ========== #
async def close_position(self, symbol, pid):
pos = self.positions[symbol][pid]
last_price = self.data[symbol]["close"].iloc[-1]
pnl = (last_price - pos.entry_price) * pos.size if pos.side == "buy" else (pos.entry_price - last_price) * pos.size
print(f"[{symbol}] CLOSE {pos.side.upper()} @ {last_price:.2f} | PnL = {pnl:.3f}")
self.cfg.account_equity += pnl
del self.positions[symbol][pid]
# ========== Main loop ========== #
async def start(self):
print("Starting multi-pair RSI bot...")
print("Symbols:", self.cfg.symbols)
while True:
try:
for symbol in self.cfg.symbols:
# new candle
ohlcv = self.get_fake_ohlcv(symbol)
self.append_candle(symbol, ohlcv)
# signal scan
await self.process_signals(symbol)
# manage trades
await self.close_expired(symbol)
except Exception as e:
print("Error:", e)
await asyncio.sleep(1)
# ========== Launch Example ========== #
async def main():
cfg = BotConfig(
symbols=["BTCUSDT", "XAUUSD", "GBPUSD"],
account_equity=3000.0,
paper=True,
lot_size=None
)
bot = MultiPairRSIBot(cfg)
task = asyncio.create_task(bot.start())
await asyncio.sleep(60 * 5) # run 5 minutes demo
task.cancel()
if __name__ == "__main__":
asyncio.run(main())
反馈
1
等级
项目
994
47%
仲裁
33
36%
/
36%
逾期
98
10%
空闲
发布者: 6 代码
2
等级
项目
23
9%
仲裁
6
33%
/
50%
逾期
1
4%
繁忙
3
等级
项目
50
28%
仲裁
14
21%
/
64%
逾期
1
2%
工作中
4
等级
项目
5
0%
仲裁
3
33%
/
67%
逾期
2
40%
空闲
5
等级
项目
11
0%
仲裁
7
29%
/
71%
逾期
2
18%
工作中
6
等级
项目
35
23%
仲裁
4
0%
/
50%
逾期
2
6%
工作中
7
等级
项目
2
0%
仲裁
2
0%
/
50%
逾期
0
空闲
8
等级
项目
2
0%
仲裁
0
逾期
0
空闲
9
等级
项目
20
10%
仲裁
8
38%
/
38%
逾期
3
15%
工作中
10
等级
项目
657
32%
仲裁
41
41%
/
46%
逾期
11
2%
繁忙
11
等级
项目
19
11%
仲裁
4
50%
/
25%
逾期
4
21%
工作中
12
等级
项目
3
33%
仲裁
2
0%
/
100%
逾期
0
空闲
13
等级
项目
3379
68%
仲裁
77
48%
/
14%
逾期
342
10%
空闲
发布者: 1 代码
14
等级
项目
3
0%
仲裁
0
逾期
0
空闲
15
等级
项目
0
0%
仲裁
1
0%
/
100%
逾期
0
空闲
16
等级
项目
1
100%
仲裁
0
逾期
0
空闲
17
等级
项目
264
30%
仲裁
0
逾期
3
1%
空闲
发布者: 2 代码
18
等级
项目
29
21%
仲裁
20
10%
/
50%
逾期
8
28%
已载入
19
等级
项目
9
0%
仲裁
2
0%
/
50%
逾期
1
11%
工作中
20
等级
项目
477
40%
仲裁
105
40%
/
24%
逾期
81
17%
已载入
发布者: 2 代码
21
等级
项目
0
0%
仲裁
0
逾期
0
空闲
22
等级
项目
0
0%
仲裁
0
逾期
0
空闲
23
等级
项目
0
0%
仲裁
0
逾期
0
空闲
相似订单
Greetings I need MT5 developer that has expertise in developing a custom indicator for mt5 boom and crash based on my exact details and requirements which would be discuss later. Kindly bid for this project if it is something you can handle for me
I need a fully functional Expert Advisor (EA) for MetaTrader 5 focused on trading XAUUSD (Gold). The strategy should be based on fast scalping entries using indicators such as RSI, Moving Averages, or similar confirmation tools. The EA must include fixed risk management (1%–2% per trade), automatic lot calculation, and stop loss/take profit settings. It should be optimized for low spread conditions and quick market
Combination of RSI and Moving averages
30 - 300 USD
I need a professional developer to help me create an EA using RSI and Moving averages.The strategies are well organised and everything is in order. I will send all the details in the inbox
MT5 Indicator SMC
30 - 100 USD
Hello, I would like to clarify something before we proceed. I already have a base code prepared for the indicator. However, the code is not the final version and it still requires professional refinement, debugging, and optimization. Because of this, I am specifically looking for a highly experienced developer who can understand the full logic and structure of the system and improve it properly. Before I send you the
Project Overview I need assistance coding a hedging Expert Advisor (EA) that operates in "closed cycles" with a strict limit of five simultaneous open positions . EA Logic & Mechanism Initial Entry: Open a Buy position of 0.01 lots (Position A) with a Take Profit (TP) target of $1.00. Scenario 1: Price Drops (Hedging Step 1): If the price moves against the first trade, open two positions simultaneously: a Sell 0.02
Hello, I would like to clarify something before we proceed. I already have a base code prepared for the indicator. However, the code is not the final version and it still requires professional refinement, debugging, and optimization. Because of this, I am specifically looking for a highly experienced developer who can understand the full logic and structure of the system and improve it properly. Before I send you the
Powerful
30 - 100 USD
I really want a powerful developed EA that can generate a minimum of 10% every month without martingale, greed or any dangerous strategy for sale. Developer must provide the mql5 file or the raw file for modification etc
I already have a fully developed MT5 Expert Advisor with all required prop firm features, including: Risk management Daily loss & max drawdown limits Spread & slippage filters News filter Trade management system The EA structure is complete. 👉 What I need is a professional developer to replace ONLY the entry logic with a high-quality, rule-based trading strategy. 🚨 STRICT REQUIREMENT (READ CAREFULLY): I am NOT
SMC Price Action Pro
30 - 100 USD
Hello, I have a source code for an MT5 indicator that is open-source and belongs to me. However, it still requires professional improvements and some corrections. At the moment the indicator does not work as expected and there are several issues in the logic and performance that need to be fixed. Because of this, I am looking for a highly experienced MT5 developer who is comfortable working with existing code
Hello, I’m looking for an experienced developer who can help convert an existing cTrader indicator into a fully functional Quantower indicator . I already have the complete source code for the cTrader indicator (written in C#) , and I would like the same logic, behavior, and visual output to be accurately replicated in Quantower
项目信息
预算
50+ USD