This page contains all project documentation, configuration, and source code for reference.
# Binance ML Spot Trading Bot
An educational automated trading bot for **Binance Spot** using a simple machine-learning model (Random Forest) to predict short-term price direction.
> ⚠️ **Warning**: This is educational software, not financial advice. Automated trading can lose money. Always start in **dry-run / testnet** mode. Never trade more than you can afford to lose.
---
## What the bot does
1. Downloads the latest candlestick data from Binance (public API).
2. Trains/loads a Random Forest model that predicts whether the next candle will close UP or DOWN.
3. Generates a **buy** signal when the model is confident price will go up.
4. Generates a **sell** signal when the model is confident price will go down, or when stop-loss / take-profit levels are hit.
5. Executes simulated trades (paper/dry-run) by default, and optionally real trades on **Binance Spot Testnet**.
6. Logs every trade to a CSV file.
7. Provides a lightweight **Streamlit dashboard** to view prices, signals, portfolio, and trades.
---
## Project structure
```
.
├── bot.py # Main trading loop
├── dashboard.py # Streamlit web dashboard
├── config.yaml # All settings (symbol, risk, ML thresholds, etc.)
├── requirements.txt # Python dependencies
├── .env.example # API keys template (only for testnet/live)
├── src/
│ ├── data.py # OHLCV download + feature engineering
│ ├── model.py # Random Forest training/prediction
│ ├── risk.py # Position sizing, stop loss, take profit
│ └── exchange.py # CCXT wrapper for Binance Spot
└── README.md
```
---
## 1. Install dependencies
```bash
cd /var/www/html/binance
python3 -m venv venv
source venv/bin/activate
pip install -r requirements.txt
```
---
## 2. Configure the bot
Open `config.yaml` and review the settings.
Key settings for beginners:
```yaml
dry_run: true # true = paper trading, no real orders
testnet: true # true = use Binance testnet (need API keys if dry_run=false)
paper_balance_usdt: 10000
symbol: BTC/USDT
timeframe: 1h
# Risk
position_size_pct: 0.10 # use 10% of USDT per trade
stop_loss_pct: 0.03 # close trade if price drops 3%
take_profit_pct: 0.06 # close trade if price rises 6%
```
Keep `dry_run: true` until you are confident the bot behaves correctly.
---
## 3. Run the bot (paper trading)
```bash
source venv/bin/activate
python bot.py
```
You will see logs like:
```
2024-01-01 12:00:00 INFO Price: 42300.50 | Signal: BUY | Up prob: 58.45%
2024-01-01 12:00:00 INFO [DRY-RUN] BUY 0.002364 BTC @ 42300.50
```
The bot will:
- Fetch a new candle every `poll_seconds` (default 5 minutes)
- Retrain the model every 24 hours
- Save trades to `trades.csv`
- Save portfolio state to `state.json`
- Save the model to `model.pkl`
Stop it anytime with `Ctrl+C`.
---
## 4. Run the dashboard
In a second terminal:
```bash
cd /var/www/html/binance
source venv/bin/activate
streamlit run dashboard.py
```
Then open the URL shown in the terminal (usually `http://localhost:8501`).
---
## 5. Move to Binance Testnet (optional, closer to real)
To test with real Binance API but **fake money**, you need Binance Spot Testnet API keys:
1. Go to https://testnet.binance.vision/
2. Generate an API key and secret.
3. Copy `.env.example` to `.env` and paste your keys:
```bash
cp .env.example .env
nano .env
```
```
BINANCE_API_KEY=your_testnet_key
BINANCE_SECRET_KEY=your_testnet_secret
```
4. Set in `config.yaml`:
```yaml
dry_run: false
testnet: true
```
5. Run the bot again. It will now place real orders on the testnet.
---
## 6. Moving to real money (NOT recommended for beginners)
Only consider this after weeks of successful dry-run/testnet results.
1. Generate real Binance API keys (with **Spot & Margin** permissions only, no withdrawal). Enable IP restrictions for safety.
2. Set in `config.yaml`:
```yaml
dry_run: false
testnet: false
```
3. Add real keys to `.env`.
4. Start with a very small amount of capital.
---
## How the ML model works (simplified)
- Features: past returns, SMA ratio, RSI, MACD, volatility, ATR, volume change.
- Target: `1` if next candle closes higher, `0` if lower.
- Model: `RandomForestClassifier` from `scikit-learn`.
- Prediction: probability that the next candle will be UP.
- If probability > `buy_threshold` → BUY.
- If probability < `sell_threshold` → SELL (or close position).
This is a basic template. ML does **not** guarantee profit.
---
## Common improvements you can add later
- Better features (order-book data, funding rates, on-chain metrics).
- Hyperparameter tuning / model selection (XGBoost, LightGBM).
- Walk-forward backtesting before going live.
- Telegram/Discord notifications.
- More advanced position sizing (Kelly, volatility targeting).
- Multi-coin portfolio.
---
## Safety checklist
- [ ] Started with `dry_run: true`?
- [ ] Watched it for several hours/days?
- [ ] Tested stop-loss and take-profit logic in paper mode?
- [ ] Only then moved to testnet with small fake balance?
- [ ] Never enabled withdrawal permissions on the API key?
---
## Questions / next steps
Tell me what you want to improve next, for example:
- Add a custom strategy (not ML).
- Add backtesting with historical data.
- Add Telegram notifications.
- Run it on a cloud server 24/7.
- Improve the ML model features.
I can help you implement any of these.
# Project Prompt: Binance ML Spot Trading Bot
## Project Overview
Build and maintain an **automated spot trading bot for Binance** that uses a simple machine-learning model to predict short-term price direction. The project is designed for a beginner who wants to learn safely, so **dry-run / paper trading must be the default**, and all real-money steps must be clearly gated behind warnings and testnet practice.
The user has:
- A Binance account (no API keys yet)
- ~10,000 QAR trading capital (use USD/USDT equivalent in paper mode)
- Preference for **Spot trading** only
- Interest in **Machine Learning** strategy
- Wants a **Streamlit dashboard**
- Wants to test on **Binance Testnet** first
---
## Tech Stack
- **Language**: Python 3.10+
- **Exchange API**: CCXT (Binance Spot + Testnet)
- **ML**: scikit-learn (`RandomForestClassifier`)
- **Data**: pandas, numpy
- **Dashboard**: Streamlit
- **Config**: YAML (`config.yaml`)
- **Secrets**: `.env` file (API keys only when leaving dry-run)
---
## Project Structure
```
.
├── bot.py # Main trading loop
├── dashboard.py # Streamlit dashboard
├── config.yaml # All user-facing settings
├── requirements.txt # Python dependencies
├── .env.example # API key template
├── .gitignore # Ignore .env, model, state, logs
├── prompt.md # This file
├── README.md # User documentation
└── src/
├── __init__.py
├── data.py # OHLCV fetching + technical features
├── model.py # Random Forest training + prediction
├── risk.py # Position sizing, stop-loss, take-profit
└── exchange.py # CCXT wrapper + dry-run paper balance
```
---
## Core Features
1. **Market Data**
- Fetch recent OHLCV candles from Binance public API (no API key needed for data).
- Build features: lagged returns, SMA ratio, RSI, MACD, volatility, ATR, volume change.
2. **Machine Learning**
- Target variable: `1` if next candle closes higher, `0` otherwise.
- Model: `RandomForestClassifier` with class balancing.
- Time-based train/test split (no shuffling of financial data).
- Retrain automatically every `retrain_interval_hours`.
- Persist trained model to `model.pkl`.
3. **Signal Generation**
- If `up_probability >= buy_threshold` → BUY.
- If `up_probability <= sell_threshold` → SELL / close position.
- Otherwise → HOLD.
4. **Risk Management**
- Position size = `balance * position_size_pct / price`.
- Minimum trade notional ~10 USDT (do not trade smaller).
- Stop-loss: close if price drops below `entry * (1 - stop_loss_pct)`.
- Take-profit: close if price rises above `entry * (1 + take_profit_pct)`.
5. **Trade Execution**
- Default mode: **dry-run / paper trading** — logs what it *would* do, no real orders.
- Optional: Binance Spot **Testnet** (requires API keys, uses fake funds).
- Optional: Binance Spot **live** (only after extensive testing, requires real API keys).
- Persist portfolio state to `state.json` so restarts remember the simulated position.
- Log all trades to `trades.csv`.
6. **Dashboard**
- Streamlit app showing: current price, latest signal, up probability, price chart, paper portfolio, trade history, total PnL.
---
## Configuration (`config.yaml`)
```yaml
symbol: BTC/USDT
timeframe: 1h
lookback: 2000
poll_seconds: 300
dry_run: true # ALWAYS true by default
testnet: true # true when dry_run=false and using Binance testnet
paper_balance_usdt: 10000
buy_threshold: 0.55
sell_threshold: 0.45
position_size_pct: 0.10
stop_loss_pct: 0.03
take_profit_pct: 0.06
retrain_interval_hours: 24
model_path: model.pkl
trades_path: trades.csv
state_path: state.json
```
---
## Safety Rules (Non-negotiable)
1. `dry_run` must be `true` in the default `config.yaml`.
2. Any real-money setup must be explicitly gated with warnings.
3. API keys must **never** have withdrawal permission.
4. API keys must be stored in `.env`, never committed (`.gitignore` must include `.env`).
5. Explain that past performance / ML accuracy does **not** guarantee future profit.
6. Before testnet: run dry-run for several hours/days and verify logs look correct.
7. Before live: run testnet for several days with small fake balance.
8. Keep code simple and commented for a beginner.
---
## How to Run
### Install
```bash
cd /var/www/html/binance
python3 -m venv venv
source venv/bin/activate
pip install -r requirements.txt
```
### Run the trading bot (paper mode)
```bash
python bot.py
```
### Run the dashboard
```bash
streamlit run dashboard.py
```
### Testnet setup (optional, fake money)
1. Get free API keys from https://testnet.binance.vision/
2. `cp .env.example .env` and paste keys.
3. Set `dry_run: false` and `testnet: true` in `config.yaml`.
4. Run `python bot.py`.
### Live setup (only after extensive testing)
1. Create real Binance Spot API keys (Spot trading permission only, no withdrawal, IP whitelist recommended).
2. Add keys to `.env`.
3. Set `dry_run: false` and `testnet: false`.
4. Start with very small capital.
---
## Common Customizations
When the user asks for changes, keep them modular and safe:
- **Change symbol/timeframe**: update `config.yaml` only.
- **Change risk**: adjust `position_size_pct`, `stop_loss_pct`, `take_profit_pct`.
- **Change ML thresholds**: adjust `buy_threshold`, `sell_threshold`.
- **Add indicators**: edit `src/data.py` and `src/model.py` `FEATURE_COLS`.
- **Try a different model**: replace `RandomForestClassifier` in `src/model.py` (XGBoost, LightGBM, etc.).
- **Add backtesting**: create a new `backtest.py` that walks through historical data without real execution.
- **Add notifications**: add Telegram/Discord webhook calls inside `bot.py` after trades.
- **Run 24/7**: provide instructions for deploying on a VPS/cloud server.
- **Add more coins**: expand to a portfolio loop (more complex, do after single-coin version is stable).
---
## Documentation Style
- Explain code simply, as if the user is a beginner.
- Use clear comments and docstrings.
- Provide copy-paste commands.
- Always include a disclaimer that this is educational, not financial advice.
---
## Disclaimer
This project is for **educational purposes only**. It is not financial advice. Trading cryptocurrencies carries significant risk, including the possibility of losing your entire capital. Always test thoroughly in dry-run and testnet modes before using real money.
# Conversation Questions
## User Questions / Requests
1. i want to create an agent to do automated trading on binance
2. اريد الناقش معك فقط
3. اريد الرد بالانجليزي اريد ان اتداول في منصه binance واريد ان اصنع اداه بوت للتداول
4. 1. Spot trading 2. ML 3. Explained simply, i have binance account bot not APIs 4. QR 10000 5. testnet 6. dashboard
5. create prompt.md for this project
6. save the all the questions in file questions.txt
## Assistant Clarifying Questions
1. What is your trading type? (Spot, Futures, or both?)
2. What strategy do you want? (SMA/EMA crossover, RSI, MACD, Bollinger Bands, Grid trading, Breakout, Machine Learning, copy signals, etc.)
3. What is your experience level with Python? Do you already have a Binance account and API keys?
4. How much capital do you plan to trade? What risk and money management rules do you want? (stop loss, take profit, risk per trade)
5. Where do you want to run the bot? (local computer, VPS, cloud server 24/7)
6. What extra features do you want? (notifications, trade logs, web dashboard, backtesting)
# Binance ML Spot Trading Bot Configuration
symbol: BTC/USDT
timeframe: 1h
lookback: 2000
poll_seconds: 300
# Safety mode: keep dry_run=true until you are confident
dry_run: true
testnet: true
paper_balance_usdt: 10000
# ML thresholds (probability that price will go up next candle)
buy_threshold: 0.55
sell_threshold: 0.45
# Risk settings
position_size_pct: 0.10 # use 10% of available USDT per trade
stop_loss_pct: 0.03 # 3% stop loss
take_profit_pct: 0.06 # 6% take profit
# Model retraining
retrain_interval_hours: 24
model_path: model.pkl
trades_path: trades.csv
state_path: state.json
ccxt>=4.0
pandas>=2.0
numpy>=1.24
scikit-learn>=1.3
streamlit>=1.28
pyyaml>=6.0
python-dotenv>=1.0
joblib>=1.3
# Only needed when dry_run=false and testnet=true
# Get free testnet keys from: https://testnet.binance.vision/
BINANCE_API_KEY=your_testnet_api_key_here
BINANCE_SECRET_KEY=your_testnet_secret_key_here
.env
venv/
__pycache__/
*.pyc
*.pkl
*.csv
*.json
.DS_Store
import logging
import os
import time
from datetime import datetime, timezone
import joblib
import pandas as pd
import yaml
from src.data import add_features, fetch_ohlcv
from src.exchange import Exchange
from src.model import predict, signal_from_prediction, train
from src.risk import RiskManager
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s %(levelname)s %(message)s",
)
logger = logging.getLogger("trading_bot")
def save_trade(trades_path, side, price, amount, reason, pnl=0.0):
"""Append a trade record to the CSV file."""
row = {
"time": datetime.now(timezone.utc).isoformat(),
"side": side,
"price": round(price, 2),
"amount": round(amount, 6),
"reason": reason,
"pnl": round(pnl, 2),
}
df = pd.DataFrame([row])
if not os.path.exists(trades_path):
df.to_csv(trades_path, index=False)
else:
df.to_csv(trades_path, mode="a", header=False, index=False)
class Bot:
def __init__(self, config):
self.config = config
self.exchange = Exchange(config)
self.risk = RiskManager(config)
self.model_path = config["model_path"]
self.trades_path = config["trades_path"]
self.model = None
self.last_train = None
def load_or_train_model(self, df):
"""Load existing model, or train a new one if too old/missing."""
if os.path.exists(self.model_path) and self.last_train is not None:
hours_since = (
datetime.now(timezone.utc) - self.last_train
).total_seconds() / 3600
if hours_since < self.config["retrain_interval_hours"]:
self.model = joblib.load(self.model_path)
return
logger.info("Training new model...")
self.model, accuracy = train(df)
logger.info(f"Model trained. Accuracy: {accuracy:.2%}")
joblib.dump(self.model, self.model_path)
self.last_train = datetime.now(timezone.utc)
def run_once(self):
df = fetch_ohlcv(
self.exchange.ccxt,
self.config["symbol"],
self.config["timeframe"],
limit=self.config["lookback"],
)
df = add_features(df)
self.load_or_train_model(df)
prob = predict(self.model, df)
signal, up_prob = signal_from_prediction(prob, self.config)
price = df["close"].iloc[-1]
logger.info(
f"Price: {price:.2f} | Signal: {signal.upper()} | Up prob: {up_prob:.2%}"
)
self._handle_signal(signal, price)
self.exchange.save_state()
def _handle_signal(self, signal, price):
if self.exchange.in_position:
sl = self.risk.stop_loss(self.exchange.entry_price)
tp = self.risk.take_profit(self.exchange.entry_price)
qty = self.exchange.quantity
if price <= sl or price >= tp:
reason = "stop_loss" if price <= sl else "take_profit"
pnl = (price - self.exchange.entry_price) * qty
self.exchange.sell(qty, price)
save_trade(self.trades_path, "SELL", price, qty, reason, pnl)
elif signal == "sell":
pnl = (price - self.exchange.entry_price) * qty
self.exchange.sell(qty, price)
save_trade(self.trades_path, "SELL", price, qty, "signal", pnl)
elif signal == "buy":
balance = self.exchange.get_balance("USDT")
amount = self.risk.position_size(balance, price)
if amount > 0:
self.exchange.buy(amount, price)
save_trade(self.trades_path, "BUY", price, amount, "signal")
def run(self):
logger.info("Bot started. Press Ctrl+C to stop.")
while True:
try:
self.run_once()
except Exception as e:
logger.error(f"Error in main loop: {e}")
time.sleep(self.config["poll_seconds"])
def main():
with open("config.yaml", "r") as f:
config = yaml.safe_load(f)
bot = Bot(config)
bot.run()
if __name__ == "__main__":
main()
import os
import sys
import ccxt
import joblib
import pandas as pd
import streamlit as st
import yaml
# Make sure we can import from the src package
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from src.data import add_features, fetch_ohlcv
from src.model import FEATURE_COLS, predict, signal_from_prediction
st.set_page_config(page_title="Binance ML Bot", layout="wide")
st.title("🤖 Binance ML Spot Trading Bot Dashboard")
# Load configuration
config = yaml.safe_load(open("config.yaml", "r"))
symbol = config["symbol"]
timeframe = config["timeframe"]
lookback = config["lookback"]
# Fetch market data
exchange = ccxt.binance(
{
"enableRateLimit": True,
"options": {"defaultType": "spot"},
}
)
with st.spinner("Loading market data..."):
df = fetch_ohlcv(exchange, symbol, timeframe, limit=lookback)
df = add_features(df)
price = df["close"].iloc[-1]
# Load or notify about model
model_path = config["model_path"]
if os.path.exists(model_path):
model = joblib.load(model_path)
prob = predict(model, df)
signal, up_prob = signal_from_prediction(prob, config)
else:
signal = "N/A"
up_prob = 0.5
st.warning(
"No trained model found. Run `python bot.py` first to train the model."
)
# Top metrics
col1, col2, col3, col4 = st.columns(4)
col1.metric("Symbol", symbol)
col2.metric("Current Price", f"${price:,.2f}")
col3.metric("Signal", signal.upper())
col4.metric("Up Probability", f"{up_prob:.2%}")
# Price chart with moving averages
st.subheader("Price Chart")
chart_df = df.set_index("timestamp")[["close", "sma_20", "sma_50"]].tail(200)
st.line_chart(chart_df)
# Portfolio (paper trading)
st.subheader("Paper Portfolio")
state_path = config["state_path"]
if os.path.exists(state_path):
state = pd.read_json(state_path, typ="series")
c1, c2, c3 = st.columns(3)
c1.metric("USDT Balance", f"${state.get('paper_balance_quote', 0):,.2f}")
c2.metric(
f"{symbol.split('/')[0]} Balance", f"{state.get('paper_balance_base', 0):.6f}"
)
c3.metric("In Position", "Yes" if state.get("in_position", False) else "No")
else:
st.info("No portfolio state yet. Start the bot to generate it.")
# Trade history
st.subheader("Trade History")
trades_path = config["trades_path"]
if os.path.exists(trades_path):
trades = pd.read_csv(trades_path)
if not trades.empty:
total_pnl = trades["pnl"].sum()
st.metric("Total PnL (USDT)", f"${total_pnl:,.2f}")
st.dataframe(trades.tail(50), use_container_width=True)
else:
st.info("No trades recorded yet.")
else:
st.info("No trades recorded yet.")
st.caption(
"⚠️ This is an educational tool. Always run in dry-run/testnet mode before using real money."
)
import pandas as pd
import numpy as np
def fetch_ohlcv(exchange, symbol, timeframe, limit=1000):
"""Download candlestick data from Binance."""
ohlcv = exchange.fetch_ohlcv(symbol, timeframe, limit=limit)
df = pd.DataFrame(
ohlcv, columns=["timestamp", "open", "high", "low", "close", "volume"]
)
df["timestamp"] = pd.to_datetime(df["timestamp"], unit="ms")
return df
def add_features(df):
"""Add technical indicators and the target variable for ML."""
df = df.copy()
# Price returns for different time lags
df["returns"] = df["close"].pct_change()
for lag in [1, 2, 3, 5, 10]:
df[f"return_lag_{lag}"] = df["returns"].shift(lag)
# Trend features
df["sma_20"] = df["close"].rolling(window=20).mean()
df["sma_50"] = df["close"].rolling(window=50).mean()
df["sma_ratio"] = df["sma_20"] / df["sma_50"] - 1.0
# Volatility
df["volatility"] = df["returns"].rolling(window=20).std()
# Momentum
df["rsi"] = compute_rsi(df["close"], 14)
df["macd"] = compute_macd(df["close"])
# Volume
df["volume_change"] = df["volume"].pct_change()
# Risk measure
df["atr"] = compute_atr(df, 14)
# Target: 1 if next candle close is higher, else 0
df["target"] = (df["close"].shift(-1) > df["close"]).astype(int)
return df.dropna()
def compute_rsi(prices, window=14):
delta = prices.diff()
gain = delta.where(delta > 0, 0).rolling(window=window).mean()
loss = (-delta.where(delta < 0, 0)).rolling(window=window).mean()
rs = gain / loss
rsi = 100 - (100 / (1 + rs))
return rsi
def compute_macd(prices, fast=12, slow=26):
ema_fast = prices.ewm(span=fast, adjust=False).mean()
ema_slow = prices.ewm(span=slow, adjust=False).mean()
return ema_fast - ema_slow
def compute_atr(df, window=14):
high_low = df["high"] - df["low"]
high_close = (df["high"] - df["close"].shift()).abs()
low_close = (df["low"] - df["close"].shift()).abs()
tr = pd.concat([high_low, high_close, low_close], axis=1).max(axis=1)
return tr.rolling(window=window).mean()
import numpy as np
from sklearn.ensemble import RandomForestClassifier
from sklearn.metrics import accuracy_score
FEATURE_COLS = [
"return_lag_1",
"return_lag_2",
"return_lag_3",
"return_lag_5",
"return_lag_10",
"sma_ratio",
"volatility",
"rsi",
"macd",
"volume_change",
"atr",
]
def train(df):
"""Train a Random Forest classifier to predict the next candle direction."""
df = df.dropna()
X = df[FEATURE_COLS]
y = df["target"]
# Time-based train/test split (do not shuffle financial data!)
split = int(len(df) * 0.8)
X_train, X_test = X.iloc[:split], X.iloc[split:]
y_train, y_test = y.iloc[:split], y.iloc[split:]
model = RandomForestClassifier(
n_estimators=200,
max_depth=6,
class_weight="balanced",
random_state=42,
n_jobs=-1,
)
model.fit(X_train, y_train)
y_pred = model.predict(X_test)
accuracy = accuracy_score(y_test, y_pred)
return model, accuracy
def predict(model, df):
"""Return the probability that the next candle will close UP."""
latest = df[FEATURE_COLS].iloc[-1:]
prob = model.predict_proba(latest)[0]
# prob[0] = down probability, prob[1] = up probability
return prob
def signal_from_prediction(prob, config):
up_prob = prob[1]
if up_prob >= config["buy_threshold"]:
return "buy", up_prob
elif up_prob <= config["sell_threshold"]:
return "sell", up_prob
return "hold", up_prob
class RiskManager:
"""Simple risk manager for position sizing, stop loss and take profit."""
def __init__(self, config):
self.position_size_pct = config["position_size_pct"]
self.stop_loss_pct = config["stop_loss_pct"]
self.take_profit_pct = config["take_profit_pct"]
def position_size(self, balance, price):
"""Return the base asset amount to buy (e.g., BTC)."""
if balance <= 0 or price <= 0:
return 0.0
quote_to_use = balance * self.position_size_pct
amount = quote_to_use / price
# Binance minimum notional is usually ~10 USDT
if quote_to_use < 10:
return 0.0
return amount
def stop_loss(self, entry_price):
return entry_price * (1 - self.stop_loss_pct)
def take_profit(self, entry_price):
return entry_price * (1 + self.take_profit_pct)
import ccxt
import json
import logging
import os
from dotenv import load_dotenv
logger = logging.getLogger("trading_bot")
class Exchange:
"""Wrapper around CCXT for Binance Spot (testnet or live)."""
def __init__(self, config):
self.config = config
self.symbol = config["symbol"]
self.dry_run = config["dry_run"]
self.use_testnet = config["testnet"]
self.state_path = config["state_path"]
self.ccxt = ccxt.binance(
{
"enableRateLimit": True,
"options": {"defaultType": "spot"},
}
)
self.ccxt.load_markets()
if not self.dry_run:
load_dotenv()
self.ccxt.apiKey = os.environ["BINANCE_API_KEY"]
self.ccxt.secret = os.environ["BINANCE_SECRET_KEY"]
if self.use_testnet:
self.ccxt.set_sandbox_mode(True)
logger.info("Connected to Binance Spot Testnet")
# Paper trading state (also used for dry_run)
self.paper_balance_quote = float(config.get("paper_balance_usdt", 10000.0))
self.paper_balance_base = 0.0
self.in_position = False
self.entry_price = 0.0
self.quantity = 0.0
self.load_state()
def fetch_ohlcv(self, symbol, timeframe, limit=1000):
return self.ccxt.fetch_ohlcv(symbol, timeframe, limit=limit)
def get_balance(self, asset="USDT"):
if self.dry_run:
quote_asset, base_asset = self.symbol.split("/")
if asset == quote_asset:
return self.paper_balance_quote
if asset == base_asset:
return self.paper_balance_base
return 0.0
balance = self.ccxt.fetch_balance()
return balance["free"].get(asset, 0.0)
def buy(self, amount, price):
"""Market buy `amount` of base asset."""
amount = self._round_amount(amount)
if amount <= 0:
return
if self.dry_run:
cost = amount * price
self.paper_balance_quote -= cost
self.paper_balance_base += amount
logger.info(f"[DRY-RUN] BUY {amount} {self.symbol.split('/')[0]} @ {price}")
else:
logger.info(f"LIVE BUY {amount} {self.symbol.split('/')[0]} @ {price}")
self.ccxt.create_market_buy_order(self.symbol, amount)
self.in_position = True
self.entry_price = price
self.quantity = amount
def sell(self, amount, price):
"""Market sell `amount` of base asset."""
amount = self._round_amount(amount)
if amount <= 0:
return
if self.dry_run:
self.paper_balance_quote += amount * price
self.paper_balance_base -= amount
logger.info(f"[DRY-RUN] SELL {amount} {self.symbol.split('/')[0]} @ {price}")
else:
logger.info(f"LIVE SELL {amount} {self.symbol.split('/')[0]} @ {price}")
self.ccxt.create_market_sell_order(self.symbol, amount)
self.in_position = False
self.entry_price = 0.0
self.quantity = 0.0
def _round_amount(self, amount):
return float(self.ccxt.amount_to_precision(self.symbol, amount))
def load_state(self):
if not os.path.exists(self.state_path):
return
try:
with open(self.state_path, "r") as f:
state = json.load(f)
self.paper_balance_quote = state.get(
"paper_balance_quote", self.paper_balance_quote
)
self.paper_balance_base = state.get("paper_balance_base", 0.0)
self.in_position = state.get("in_position", False)
self.entry_price = state.get("entry_price", 0.0)
self.quantity = state.get("quantity", 0.0)
except Exception as e:
logger.warning(f"Could not load state: {e}")
def save_state(self):
state = {
"paper_balance_quote": self.paper_balance_quote,
"paper_balance_base": self.paper_balance_base,
"in_position": self.in_position,
"entry_price": self.entry_price,
"quantity": self.quantity,
}
with open(self.state_path, "w") as f:
json.dump(state, f, indent=2)