885 lines
32 KiB
Python
885 lines
32 KiB
Python
import hashlib
|
|
import hmac
|
|
import json
|
|
import time
|
|
from typing import Dict, Optional, List
|
|
from urllib.parse import urlencode
|
|
|
|
import requests
|
|
from .config import get_config
|
|
|
|
import json
|
|
from datetime import datetime, timedelta, timezone
|
|
import pandas as pd
|
|
from stockstats import StockDataFrame
|
|
|
|
|
|
def bybit_v5_request(method: str, path: str, params: Optional[Dict] = None, body: Optional[Dict] = None) -> Dict:
|
|
"""Generic signed HTTP request helper for Bybit V5 API."""
|
|
config = get_config()["external"]
|
|
base_url = config["BYBIT_BASE_URL"].rstrip("/")
|
|
api_key = config["BYBIT_API_KEY"]
|
|
api_secret = config["BYBIT_API_SECRET"]
|
|
|
|
if not api_key or not api_secret:
|
|
raise ValueError("Missing BYBIT_API_KEY or BYBIT_API_SECRET")
|
|
|
|
timestamp = str(int(time.time() * 1000))
|
|
recv_window = "5000"
|
|
|
|
# Build query string for GET or body for POST
|
|
if method.upper() == "GET" and params:
|
|
query_string = urlencode(sorted(params.items()))
|
|
url = f"{base_url}{path}?{query_string}"
|
|
payload = query_string
|
|
else:
|
|
url = f"{base_url}{path}"
|
|
payload = json.dumps(body, separators=(',', ':')) if body else ""
|
|
|
|
# Create signature
|
|
sign_payload = f"{timestamp}{api_key}{recv_window}{payload}"
|
|
signature = hmac.new(
|
|
api_secret.encode('utf-8'),
|
|
sign_payload.encode('utf-8'),
|
|
hashlib.sha256
|
|
).hexdigest()
|
|
|
|
# Headers
|
|
headers = {
|
|
"X-BAPI-API-KEY": api_key,
|
|
"X-BAPI-TIMESTAMP": timestamp,
|
|
"X-BAPI-RECV-WINDOW": recv_window,
|
|
"X-BAPI-SIGN": signature,
|
|
"Content-Type": "application/json"
|
|
}
|
|
|
|
# Make request
|
|
if method.upper() == "GET":
|
|
response = requests.get(url, headers=headers)
|
|
else:
|
|
response = requests.post(url, headers=headers, data=payload)
|
|
|
|
response.raise_for_status()
|
|
data = response.json()
|
|
|
|
if data.get("retCode") != 0:
|
|
raise ValueError(f"Bybit API error: {data.get('retMsg')}")
|
|
|
|
return data
|
|
|
|
def get_account_balance(symbol: str) -> dict:
|
|
"""
|
|
To determine total equity, available free margin for new trades, and locked capital.
|
|
|
|
Args:
|
|
symbol: Format "BASE/QUOTE" (e.g., "BTC/USDT")
|
|
quote_coin: The currency used for buying (e.g., "USDT")
|
|
"""
|
|
if "/" not in symbol:
|
|
return f"Error: Symbol '{symbol}' is not in the correct format. Please use 'BASE/QUOTE' format, e.g., 'BTC/USDT'."
|
|
base_coin, quote_coin = symbol.split("/")
|
|
# 1. Fetch all assets from Bybit (omitting 'coin' gets everything)
|
|
data = bybit_v5_request("GET", "/v5/account/wallet-balance", {
|
|
"accountType": "UNIFIED"
|
|
})
|
|
|
|
|
|
# 2. Parse the raw response
|
|
try:
|
|
raw_list = data["result"]["list"][0]["coin"]
|
|
# Convert list to a dictionary for easy lookup: {'BTC': {...}, 'USDT': {...}}
|
|
result = {
|
|
item["coin"]: {k: float(v) if v else 0.0 for k, v in item.items() if k != "coin"}
|
|
for item in raw_list
|
|
}
|
|
except (IndexError, KeyError, TypeError):
|
|
result = {"error": "Could not retrieve wallet balance"}
|
|
|
|
total_equity = sum(asset.get("usdValue", 0.0) for asset in result.values())
|
|
|
|
report = f"# Account Balance Report for {base_coin}/{quote_coin}\n"
|
|
report += f"** Total Equity: ${total_equity} **\n"
|
|
report += f"## {quote_coin} (Quote) Details:\n"
|
|
report += json.dumps(result.get(quote_coin, {}), indent=2) + "\n"
|
|
report += f"## {base_coin} (Base) Details:\n"
|
|
report += json.dumps(result.get(base_coin, {}), indent=2) + "\n"
|
|
return report
|
|
|
|
def get_symbol(base_coin: str, quote_coin: str) -> str:
|
|
"""
|
|
Safely retrieves the correct Bybit symbol (e.g., "BTCUSDT") for a given base/quote pair.
|
|
|
|
Args:
|
|
base_coin: The asset (e.g., "BTC")
|
|
quote_coin: The currency (e.g., "USDT")
|
|
category: "linear", "spot", or "inverse"
|
|
|
|
Returns:
|
|
The valid symbol string (e.g., "BTCUSDT") or None if not found.
|
|
"""
|
|
# 1. Query the API specifically for this Base Coin
|
|
# This filters the search on the server side, which is much faster.
|
|
params = {
|
|
"category": "spot",
|
|
"baseCoin": base_coin.upper(),
|
|
"limit": 20 # We only expect a few matches (e.g., BTCUSDT, BTC-PERP)
|
|
}
|
|
|
|
data = bybit_v5_request("GET", "/v5/market/instruments-info", params)
|
|
|
|
result = data.get("result", {})
|
|
instruments = result.get("list", [])
|
|
|
|
# 2. Find the exact match for the Quote Coin
|
|
# This handles cases where BTC might pair with USDT, USDC, or DAI
|
|
for item in instruments:
|
|
if item.get("quoteCoin") == quote_coin.upper() and item.get("baseCoin") == base_coin.upper():
|
|
return item.get("symbol")
|
|
|
|
# 3. Fallback/Error handling
|
|
return None
|
|
|
|
def get_open_orders(symbol: str) -> str:
|
|
"""
|
|
Fetches active orders and returns a text report analyzing capital lock-up and order age.
|
|
"""
|
|
if "/" not in symbol:
|
|
return f"Error: Symbol '{symbol}' is not in the correct format. Please use 'BASE/QUOTE' format, e.g., 'BTC/USDT'."
|
|
base_coin, quote_coin = symbol.split("/")
|
|
symbol = get_symbol(base_coin, quote_coin)
|
|
if not symbol:
|
|
return f"Error: No valid spot symbol found for {base_coin}/{quote_coin}"
|
|
# 1. Fetch Open Orders
|
|
data = bybit_v5_request("GET", "/v5/order/realtime", {
|
|
"category": "spot",
|
|
"symbol": symbol.upper(),
|
|
"openOnly": 0 # 0=Active orders (Pending)
|
|
})
|
|
|
|
result = data.get("result", {})
|
|
orders = result.get("list", [])
|
|
|
|
for i in range(len(orders)):
|
|
# try to change to float if can
|
|
for k, v in orders[i].items():
|
|
if k in ["orderLinkId", "orderId"]:
|
|
continue
|
|
try:
|
|
orders[i][k] = float(v)
|
|
except:
|
|
pass
|
|
# change createdTime and updatedTime to yyyy-mm-dd hh:mm:ss format (utc)
|
|
orders[i]["createdTime"] = time.strftime('%Y-%m-%d %H:%M:%S', time.gmtime(orders[i]["createdTime"]/1000))
|
|
orders[i]["updatedTime"] = time.strftime('%Y-%m-%d %H:%M:%S', time.gmtime(orders[i]["updatedTime"]/1000))
|
|
|
|
report = f"# Open Orders for {symbol.upper()}\n"
|
|
report += json.dumps(orders, indent=2)
|
|
|
|
return report
|
|
|
|
|
|
def get_market_data(symbol:str, start_date: str, end_date: str) -> str:
|
|
"""
|
|
Fetches historical Daily (1D) OHLCV data for a specific date range.
|
|
|
|
Args:
|
|
symbol: Format "BASE/QUOTE" (e.g., "BTC/USDT").
|
|
start_date: Format "YYYY-MM-DD" (Inclusive).
|
|
end_date: Format "YYYY-MM-DD" (Inclusive).
|
|
|
|
Returns:
|
|
A formatted CSV-style string report.
|
|
"""
|
|
if "/" not in symbol:
|
|
return f"Error: Symbol '{symbol}' is not in the correct format. Please use 'BASE/QUOTE' format, e.g., 'BTC/USDT'."
|
|
# 1. Convert Date Strings to Timestamps (ms)
|
|
try:
|
|
# Start of the start_date (00:00:00)
|
|
dt_start = datetime.strptime(start_date, "%Y-%m-%d")
|
|
ts_start = int(dt_start.timestamp() * 1000)
|
|
|
|
# End of the end_date (23:59:59) - Bybit 'end' parameter is inclusive if valid candle exists
|
|
dt_end = datetime.strptime(end_date, "%Y-%m-%d") + timedelta(days=1)
|
|
ts_end = int(dt_end.timestamp() * 1000)
|
|
except ValueError:
|
|
return "Error: Invalid date format. Please use YYYY-MM-DD."
|
|
|
|
# 2. Request Data from Bybit
|
|
# We use interval "D" for Daily. Limit 1000 covers ~2.7 years of daily data in one call.
|
|
base_coin, quote_coin = symbol.split("/")
|
|
symbol2 = get_symbol(base_coin, quote_coin)
|
|
if not symbol2:
|
|
return f"# Error: No valid spot symbol found for {base_coin}/{quote_coin}."
|
|
params = {
|
|
"category": "spot",
|
|
"symbol": symbol2.upper(),
|
|
"interval": "D",
|
|
"start": ts_start,
|
|
"end": ts_end,
|
|
"limit": 1000
|
|
}
|
|
|
|
data = bybit_v5_request("GET", "/v5/market/kline", params)
|
|
|
|
result = data.get("result", {})
|
|
raw_candles = result.get("list", []) # Returns [timestamp, open, high, low, close, volume, turnover]
|
|
|
|
if not raw_candles:
|
|
return f"# No market data found for {symbol.upper()} from {start_date} to {end_date}."
|
|
|
|
# 3. Format Data
|
|
# Bybit returns Newest -> Oldest. We reverse it to get chronological order (Oldest -> Newest).
|
|
raw_candles.reverse()
|
|
|
|
csv_lines = []
|
|
|
|
for candle in raw_candles:
|
|
# Parse Timestamp
|
|
ts_ms = int(candle[0])
|
|
date_str = datetime.fromtimestamp(ts_ms / 1000, tz=timezone.utc).strftime('%Y-%m-%d')
|
|
|
|
# Parse Values
|
|
open_p = candle[1]
|
|
high_p = candle[2]
|
|
low_p = candle[3]
|
|
close_p = candle[4]
|
|
volume = float(candle[5]) # Volume in Base Currency (e.g. BTC)
|
|
|
|
# Format Line: Date,Open,High,Low,Close,Volume
|
|
# Note: We omit Dividends/Stock Splits as requested
|
|
line = f"{date_str},{open_p},{high_p},{low_p},{close_p},{int(volume)}"
|
|
csv_lines.append(line)
|
|
|
|
# 4. Construct Final Report Header
|
|
current_time = datetime.now(tz=timezone.utc).strftime('%Y-%m-%d %H:%M:%S UTC')
|
|
|
|
header = [
|
|
f"# Crypto data for {symbol.upper()} from {start_date} to {end_date}",
|
|
f"# Total records: {len(csv_lines)}",
|
|
f"# Data retrieved on: {current_time}",
|
|
"",
|
|
"Date,Open,High,Low,Close,Volume"
|
|
]
|
|
|
|
return "\n".join(header + csv_lines)
|
|
|
|
def get_crypto_indicator_window(
|
|
symbol: str,
|
|
indicator: str,
|
|
curr_date: str,
|
|
look_back_days: int
|
|
) -> str:
|
|
"""
|
|
Calculates technical indicators for a crypto pair using Bybit data.
|
|
Fix: Uses index lookup for dates since stockstats sets 'date' as the index.
|
|
"""
|
|
if "/" not in symbol:
|
|
return f"Error: Symbol '{symbol}' is not in the correct format. Please use 'BASE/QUOTE' format, e.g., 'BTC/USDT'."
|
|
base_coin, quote_coin = symbol.split("/")
|
|
symbol2 = get_symbol(base_coin, quote_coin)
|
|
|
|
# 1. Define Supported Indicators (Same as before)
|
|
best_ind_params = {
|
|
"close_50_sma": (
|
|
"50 SMA: A medium-term trend indicator. "
|
|
"Usage: Identify trend direction and serve as dynamic support/resistance. "
|
|
"Tips: It lags price; combine with faster indicators for timely signals."
|
|
),
|
|
"close_200_sma": (
|
|
"200 SMA: A long-term trend benchmark. "
|
|
"Usage: Confirm overall market trend and identify golden/death cross setups. "
|
|
"Tips: It reacts slowly; best for strategic trend confirmation rather than frequent trading entries."
|
|
),
|
|
"close_10_ema": (
|
|
"10 EMA: A responsive short-term average. "
|
|
"Usage: Capture quick shifts in momentum and potential entry points. "
|
|
"Tips: Prone to noise in choppy markets; use alongside longer averages for filtering false signals."
|
|
),
|
|
# MACD Related
|
|
"macd": (
|
|
"MACD: Computes momentum via differences of EMAs. "
|
|
"Usage: Look for crossovers and divergence as signals of trend changes. "
|
|
"Tips: Confirm with other indicators in low-volatility or sideways markets."
|
|
),
|
|
"macds": (
|
|
"MACD Signal: An EMA smoothing of the MACD line. "
|
|
"Usage: Use crossovers with the MACD line to trigger trades. "
|
|
"Tips: Should be part of a broader strategy to avoid false positives."
|
|
),
|
|
"macdh": (
|
|
"MACD Histogram: Shows the gap between the MACD line and its signal. "
|
|
"Usage: Visualize momentum strength and spot divergence early. "
|
|
"Tips: Can be volatile; complement with additional filters in fast-moving markets."
|
|
),
|
|
# Momentum Indicators
|
|
"rsi": (
|
|
"RSI: Measures momentum to flag overbought/oversold conditions. "
|
|
"Usage: Apply 70/30 thresholds and watch for divergence to signal reversals. "
|
|
"Tips: In strong trends, RSI may remain extreme; always cross-check with trend analysis."
|
|
),
|
|
# Volatility Indicators
|
|
"boll": (
|
|
"Bollinger Middle: A 20 SMA serving as the basis for Bollinger Bands. "
|
|
"Usage: Acts as a dynamic benchmark for price movement. "
|
|
"Tips: Combine with the upper and lower bands to effectively spot breakouts or reversals."
|
|
),
|
|
"boll_ub": (
|
|
"Bollinger Upper Band: Typically 2 standard deviations above the middle line. "
|
|
"Usage: Signals potential overbought conditions and breakout zones. "
|
|
"Tips: Confirm signals with other tools; prices may ride the band in strong trends."
|
|
),
|
|
"boll_lb": (
|
|
"Bollinger Lower Band: Typically 2 standard deviations below the middle line. "
|
|
"Usage: Indicates potential oversold conditions. "
|
|
"Tips: Use additional analysis to avoid false reversal signals."
|
|
),
|
|
"atr": (
|
|
"ATR: Averages true range to measure volatility. "
|
|
"Usage: Set stop-loss levels and adjust position sizes based on current market volatility. "
|
|
"Tips: It's a reactive measure, so use it as part of a broader risk management strategy."
|
|
),
|
|
# Volume-Based Indicators
|
|
"vwma": (
|
|
"VWMA: A moving average weighted by volume. "
|
|
"Usage: Confirm trends by integrating price action with volume data. "
|
|
"Tips: Watch for skewed results from volume spikes; use in combination with other volume analyses."
|
|
),
|
|
"mfi": (
|
|
"MFI: The Money Flow Index is a momentum indicator that uses both price and volume to measure buying and selling pressure. "
|
|
"Usage: Identify overbought (>80) or oversold (<20) conditions and confirm the strength of trends or reversals. "
|
|
"Tips: Use alongside RSI or MACD to confirm signals; divergence between price and MFI can indicate potential reversals."
|
|
),
|
|
}
|
|
|
|
# 2. Calculate Date Range
|
|
target_date_dt = datetime.strptime(curr_date, "%Y-%m-%d")
|
|
start_window_dt = target_date_dt - timedelta(days=look_back_days)
|
|
|
|
# Buffer: Fetch 250 extra days before start_window for lagging indicators (like 200 SMA)
|
|
buffer_days = 250
|
|
fetch_start_dt = start_window_dt - timedelta(days=buffer_days)
|
|
|
|
# Convert to timestamps for Bybit (ms)
|
|
ts_start = int(fetch_start_dt.timestamp() * 1000)
|
|
ts_end = int((target_date_dt + timedelta(days=1)).timestamp() * 1000)
|
|
|
|
# 3. Fetch Data from Bybit
|
|
params = {
|
|
"category": "linear",
|
|
"symbol": symbol2.upper(),
|
|
"interval": "D",
|
|
"start": ts_start,
|
|
"end": ts_end,
|
|
"limit": 1000
|
|
}
|
|
|
|
# Ensure 'bybit_v5_request' is defined in your scope
|
|
data = bybit_v5_request("GET", "/v5/market/kline", params)
|
|
raw_list = data.get("result", {}).get("list", [])
|
|
|
|
if not raw_list:
|
|
return f"Error: No data found for {symbol2}."
|
|
|
|
# 4. Prepare DataFrame
|
|
parsed_data = []
|
|
for row in raw_list:
|
|
parsed_data.append({
|
|
# Standardize date format for the Index
|
|
"date": datetime.fromtimestamp(int(row[0])/1000).strftime('%Y-%m-%d'),
|
|
"open": float(row[1]),
|
|
"high": float(row[2]),
|
|
"low": float(row[3]),
|
|
"close": float(row[4]),
|
|
"volume": float(row[5])
|
|
})
|
|
|
|
df = pd.DataFrame(parsed_data)
|
|
# Sort Ascending (Oldest -> Newest) is CRITICAL for stockstats
|
|
df = df.sort_values("date").reset_index(drop=True)
|
|
|
|
# 5. Calculate Indicator
|
|
# retype(df) sets 'date' column as the Index!
|
|
stock = StockDataFrame.retype(df)
|
|
|
|
try:
|
|
# Trigger calculation
|
|
_ = stock[indicator]
|
|
except KeyError:
|
|
return f"Error: Could not calculate {indicator}. Check if supported."
|
|
|
|
# 6. Build Report
|
|
report_lines = []
|
|
current_check_date = target_date_dt
|
|
|
|
while current_check_date >= start_window_dt:
|
|
date_str = current_check_date.strftime('%Y-%m-%d')
|
|
|
|
# --- FIX: Use .loc to find the row by Index (Date) ---
|
|
try:
|
|
# Locate the row using the date string index
|
|
val = stock.loc[date_str][indicator]
|
|
|
|
# Format value
|
|
if isinstance(val, (float, int)):
|
|
val_str = f"{val:.4f}"
|
|
else:
|
|
val_str = str(val)
|
|
|
|
report_lines.append(f"{date_str}: {val_str}")
|
|
|
|
except KeyError:
|
|
# Date not found in index
|
|
report_lines.append(f"{date_str}: N/A (No Data)")
|
|
except Exception as e:
|
|
report_lines.append(f"{date_str}: Error ({str(e)})")
|
|
|
|
current_check_date -= timedelta(days=1)
|
|
|
|
# 7. Final Output
|
|
description = best_ind_params.get(indicator, "No description available.")
|
|
|
|
result_str = (
|
|
f"## {indicator} values for {symbol2} from {start_window_dt.strftime('%Y-%m-%d')} to {curr_date}:\n\n"
|
|
+ "\n".join(report_lines)
|
|
+ "\n\n"
|
|
+ description
|
|
)
|
|
|
|
return result_str
|
|
|
|
def get_crypto_indicators_bulk(
|
|
symbol: str,
|
|
indicators: List[str],
|
|
curr_date: str,
|
|
look_back_days: int
|
|
) -> str:
|
|
"""
|
|
Calculates multiple technical indicators for a crypto pair in one go.
|
|
|
|
Args:
|
|
symbol: e.g., "BTC/USDT"
|
|
indicators: List of keys, e.g. ["rsi", "close_200_sma", "macd"]
|
|
curr_date: "YYYY-MM-DD"
|
|
look_back_days: Days of history to show in the report.
|
|
"""
|
|
if "/" not in symbol:
|
|
return f"Error: Symbol '{symbol}' is not in the correct format. Please use 'BASE/QUOTE' format, e.g., 'BTC/USDT'."
|
|
base_coin, quote_coin = symbol.split("/")
|
|
symbol2 = get_symbol(base_coin, quote_coin)
|
|
|
|
# 1. Define Descriptions
|
|
best_ind_params = {
|
|
"close_50_sma": (
|
|
"50 SMA: A medium-term trend indicator. "
|
|
"Usage: Identify trend direction and serve as dynamic support/resistance. "
|
|
"Tips: It lags price; combine with faster indicators for timely signals."
|
|
),
|
|
"close_200_sma": (
|
|
"200 SMA: A long-term trend benchmark. "
|
|
"Usage: Confirm overall market trend and identify golden/death cross setups. "
|
|
"Tips: It reacts slowly; best for strategic trend confirmation rather than frequent trading entries."
|
|
),
|
|
"close_10_ema": (
|
|
"10 EMA: A responsive short-term average. "
|
|
"Usage: Capture quick shifts in momentum and potential entry points. "
|
|
"Tips: Prone to noise in choppy markets; use alongside longer averages for filtering false signals."
|
|
),
|
|
# MACD Related
|
|
"macd": (
|
|
"MACD: Computes momentum via differences of EMAs. "
|
|
"Usage: Look for crossovers and divergence as signals of trend changes. "
|
|
"Tips: Confirm with other indicators in low-volatility or sideways markets."
|
|
),
|
|
"macds": (
|
|
"MACD Signal: An EMA smoothing of the MACD line. "
|
|
"Usage: Use crossovers with the MACD line to trigger trades. "
|
|
"Tips: Should be part of a broader strategy to avoid false positives."
|
|
),
|
|
"macdh": (
|
|
"MACD Histogram: Shows the gap between the MACD line and its signal. "
|
|
"Usage: Visualize momentum strength and spot divergence early. "
|
|
"Tips: Can be volatile; complement with additional filters in fast-moving markets."
|
|
),
|
|
# Momentum Indicators
|
|
"rsi": (
|
|
"RSI: Measures momentum to flag overbought/oversold conditions. "
|
|
"Usage: Apply 70/30 thresholds and watch for divergence to signal reversals. "
|
|
"Tips: In strong trends, RSI may remain extreme; always cross-check with trend analysis."
|
|
),
|
|
# Volatility Indicators
|
|
"boll": (
|
|
"Bollinger Middle: A 20 SMA serving as the basis for Bollinger Bands. "
|
|
"Usage: Acts as a dynamic benchmark for price movement. "
|
|
"Tips: Combine with the upper and lower bands to effectively spot breakouts or reversals."
|
|
),
|
|
"boll_ub": (
|
|
"Bollinger Upper Band: Typically 2 standard deviations above the middle line. "
|
|
"Usage: Signals potential overbought conditions and breakout zones. "
|
|
"Tips: Confirm signals with other tools; prices may ride the band in strong trends."
|
|
),
|
|
"boll_lb": (
|
|
"Bollinger Lower Band: Typically 2 standard deviations below the middle line. "
|
|
"Usage: Indicates potential oversold conditions. "
|
|
"Tips: Use additional analysis to avoid false reversal signals."
|
|
),
|
|
"atr": (
|
|
"ATR: Averages true range to measure volatility. "
|
|
"Usage: Set stop-loss levels and adjust position sizes based on current market volatility. "
|
|
"Tips: It's a reactive measure, so use it as part of a broader risk management strategy."
|
|
),
|
|
# Volume-Based Indicators
|
|
"vwma": (
|
|
"VWMA: A moving average weighted by volume. "
|
|
"Usage: Confirm trends by integrating price action with volume data. "
|
|
"Tips: Watch for skewed results from volume spikes; use in combination with other volume analyses."
|
|
),
|
|
"mfi": (
|
|
"MFI: The Money Flow Index is a momentum indicator that uses both price and volume to measure buying and selling pressure. "
|
|
"Usage: Identify overbought (>80) or oversold (<20) conditions and confirm the strength of trends or reversals. "
|
|
"Tips: Use alongside RSI or MACD to confirm signals; divergence between price and MFI can indicate potential reversals."
|
|
),
|
|
}
|
|
|
|
# 2. Fetch Data (ONLY ONCE)
|
|
# ---------------------------------------------------------
|
|
target_date_dt = datetime.strptime(curr_date, "%Y-%m-%d")
|
|
start_window_dt = target_date_dt - timedelta(days=look_back_days)
|
|
|
|
# Buffer for lagging indicators
|
|
buffer_days = 250
|
|
fetch_start_dt = start_window_dt - timedelta(days=buffer_days)
|
|
|
|
ts_start = int(fetch_start_dt.timestamp() * 1000)
|
|
ts_end = int((target_date_dt + timedelta(days=1)).timestamp() * 1000)
|
|
|
|
params = {
|
|
"category": "linear",
|
|
"symbol": symbol2.upper(),
|
|
"interval": "D",
|
|
"start": ts_start,
|
|
"end": ts_end,
|
|
"limit": 1000
|
|
}
|
|
|
|
data = bybit_v5_request("GET", "/v5/market/kline", params)
|
|
raw_list = data.get("result", {}).get("list", [])
|
|
|
|
if not raw_list:
|
|
return f"Error: No data found for {symbol2}."
|
|
|
|
parsed_data = []
|
|
for row in raw_list:
|
|
parsed_data.append({
|
|
"date": datetime.fromtimestamp(int(row[0])/1000).strftime('%Y-%m-%d'),
|
|
"open": float(row[1]),
|
|
"high": float(row[2]),
|
|
"low": float(row[3]),
|
|
"close": float(row[4]),
|
|
"volume": float(row[5])
|
|
})
|
|
|
|
df = pd.DataFrame(parsed_data)
|
|
df = df.sort_values("date").reset_index(drop=True)
|
|
stock = StockDataFrame.retype(df)
|
|
# ---------------------------------------------------------
|
|
|
|
# 3. Process Each Indicator
|
|
final_report = [f"# Technical Indicators Report for {symbol2}\n" + '-' * 40]
|
|
|
|
for ind in indicators:
|
|
# Pre-calculation check
|
|
try:
|
|
# Trigger calculation of the specific column
|
|
_ = stock[ind]
|
|
except (KeyError, ValueError):
|
|
final_report.append(f"## Error: Indicator '{ind}' is not supported or failed to calculate.\n")
|
|
continue
|
|
|
|
# Generate Block Report
|
|
report_lines = []
|
|
current_check_date = target_date_dt
|
|
|
|
while current_check_date >= start_window_dt:
|
|
date_str = current_check_date.strftime('%Y-%m-%d')
|
|
try:
|
|
# Use .loc because 'date' is the index
|
|
val = stock.loc[date_str][ind]
|
|
|
|
if isinstance(val, (float, int)):
|
|
val_str = f"{val:.4f}"
|
|
else:
|
|
val_str = str(val)
|
|
report_lines.append(f"{date_str}: {val_str}")
|
|
except KeyError:
|
|
report_lines.append(f"{date_str}: N/A")
|
|
|
|
current_check_date -= timedelta(days=1)
|
|
|
|
description = best_ind_params.get(ind, "No description available.")
|
|
|
|
block = (
|
|
f"## {ind} values for {symbol2} from {start_window_dt.strftime('%Y-%m-%d')} to {curr_date}:\n\n"
|
|
+ "\n".join(report_lines)
|
|
+ "\n\n"
|
|
+ f"Description: {description}\n"
|
|
+ "-" * 40 # Separator line
|
|
)
|
|
final_report.append(block)
|
|
|
|
return "\n\n".join(final_report)
|
|
|
|
def get_order_status(order_id: str, category: str = "spot") -> Dict:
|
|
"""
|
|
Get order status by order ID.
|
|
|
|
Args:
|
|
order_id: Order ID to query
|
|
category: Trading category ("spot", "linear", "inverse")
|
|
|
|
Returns:
|
|
Dict containing order information
|
|
"""
|
|
params = {
|
|
"category": category.lower(),
|
|
"orderId": order_id
|
|
}
|
|
|
|
data = bybit_v5_request("GET", "/v5/order/realtime", params=params)
|
|
result = data.get("result", {})
|
|
orders = result.get("list", [])
|
|
|
|
return orders[0] if orders else {}
|
|
|
|
|
|
def cancel_order(order_id: str, symbol: str, category: str = "spot") -> Dict:
|
|
"""
|
|
Cancel an existing order.
|
|
|
|
Args:
|
|
order_id: Order ID to cancel
|
|
symbol: Trading pair symbol
|
|
category: Trading category ("spot", "linear", "inverse")
|
|
|
|
Returns:
|
|
Dict containing cancellation result
|
|
"""
|
|
body = {
|
|
"category": category.lower(),
|
|
"symbol": symbol.upper(),
|
|
"orderId": order_id
|
|
}
|
|
|
|
data = bybit_v5_request("POST", "/v5/order/cancel", body=body)
|
|
return data.get("result", {})
|
|
|
|
|
|
def get_order_history(
|
|
symbol: Optional[str] = None,
|
|
category: str = "spot",
|
|
limit: int = 20
|
|
) -> Dict:
|
|
"""
|
|
Get order history.
|
|
|
|
Args:
|
|
symbol: Optional trading pair to filter by
|
|
category: Trading category ("spot", "linear", "inverse")
|
|
limit: Number of records to return (max 50)
|
|
|
|
Returns:
|
|
Dict containing order history
|
|
"""
|
|
params = {
|
|
"category": category.lower(),
|
|
"limit": min(limit, 50)
|
|
}
|
|
|
|
if symbol:
|
|
params["symbol"] = symbol.upper()
|
|
|
|
data = bybit_v5_request("GET", "/v5/order/history", params=params)
|
|
return data.get("result", {})
|
|
|
|
|
|
def get_account_info(account_type: str = "UNIFIED") -> Dict:
|
|
"""
|
|
Get account information.
|
|
|
|
Args:
|
|
account_type: Account type ("UNIFIED")
|
|
|
|
Returns:
|
|
Dict containing account information
|
|
"""
|
|
params = {
|
|
"accountType": account_type
|
|
}
|
|
|
|
data = bybit_v5_request("GET", "/v5/account/info", params=params)
|
|
return data.get("result", {})
|
|
|
|
def place_order(
|
|
symbol: str,
|
|
side: str,
|
|
order_type: str,
|
|
qty: float,
|
|
price: Optional[float] = None,
|
|
stop_loss: Optional[float] = None,
|
|
take_profit: Optional[float] = None,
|
|
sl_limit_price: Optional[float] = None,
|
|
tp_limit_price: Optional[float] = None,
|
|
sl_order_type: str = "Market",
|
|
tp_order_type: str = "Market",
|
|
time_in_force: str = "GTC",
|
|
account_type: str = "UNIFIED",
|
|
category: str = "spot",
|
|
order_link_id: Optional[str] = None,
|
|
reduce_only: bool = False,
|
|
close_on_trigger: bool = False
|
|
) -> Dict:
|
|
"""
|
|
Place an order on Bybit with comprehensive support for spot trading.
|
|
|
|
Args:
|
|
symbol: Trading pair symbol (e.g., "BTCUSDT")
|
|
side: Order side ("Buy" or "Sell")
|
|
order_type: Order type ("Market", "Limit")
|
|
qty: Order quantity
|
|
price: Order price (required for Limit orders)
|
|
stop_loss: Stop loss trigger price
|
|
take_profit: Take profit trigger price
|
|
sl_limit_price: Stop loss limit price (for limit SL orders)
|
|
tp_limit_price: Take profit limit price (for limit TP orders)
|
|
sl_order_type: Stop loss order type ("Market" or "Limit")
|
|
tp_order_type: Take profit order type ("Market" or "Limit")
|
|
time_in_force: Time in force ("GTC", "IOC", "FOK", "PostOnly")
|
|
account_type: Account type ("UNIFIED")
|
|
category: Trading category ("spot", "linear", "inverse")
|
|
order_link_id: Custom order ID
|
|
reduce_only: Reduce only flag
|
|
close_on_trigger: Close on trigger flag
|
|
|
|
Returns:
|
|
Dict containing order result
|
|
|
|
Raises:
|
|
ValueError: If required parameters are missing or invalid
|
|
"""
|
|
# Validate required parameters
|
|
if not symbol or not side or not order_type:
|
|
raise ValueError("symbol, side, and order_type are required")
|
|
|
|
if qty <= 0:
|
|
raise ValueError("qty must be greater than 0")
|
|
|
|
# Validate order type and price requirement
|
|
if order_type.upper() == "LIMIT" and price is None:
|
|
raise ValueError("price is required for Limit orders")
|
|
|
|
# Validate side
|
|
if side.upper() not in ["BUY", "SELL"]:
|
|
raise ValueError("side must be 'Buy' or 'Sell'")
|
|
|
|
# Validate order type
|
|
valid_order_types = ["MARKET", "LIMIT"]
|
|
if order_type.upper() not in valid_order_types:
|
|
raise ValueError(f"order_type must be one of {valid_order_types}")
|
|
|
|
# Build order body
|
|
body = {
|
|
"category": category.lower(),
|
|
"symbol": symbol.upper(),
|
|
"side": side.capitalize(),
|
|
"orderType": order_type.capitalize(),
|
|
"qty": str(qty),
|
|
"timeInForce": time_in_force,
|
|
}
|
|
|
|
# Add price for limit orders
|
|
if price is not None:
|
|
body["price"] = str(price)
|
|
|
|
# Add stop loss with proper formatting
|
|
if stop_loss is not None:
|
|
body["stopLoss"] = str(stop_loss)
|
|
body["slOrderType"] = sl_order_type.capitalize()
|
|
|
|
# Add limit price for stop loss if specified
|
|
if sl_order_type.upper() == "LIMIT" and sl_limit_price is not None:
|
|
body["slLimitPrice"] = str(sl_limit_price)
|
|
|
|
# Add take profit with proper formatting
|
|
if take_profit is not None:
|
|
body["takeProfit"] = str(take_profit)
|
|
body["tpOrderType"] = tp_order_type.capitalize()
|
|
|
|
# Add limit price for take profit if specified
|
|
if tp_order_type.upper() == "LIMIT" and tp_limit_price is not None:
|
|
body["tpLimitPrice"] = str(tp_limit_price)
|
|
|
|
# Add optional parameters
|
|
if order_link_id:
|
|
body["orderLinkId"] = order_link_id
|
|
|
|
if reduce_only:
|
|
body["reduceOnly"] = True
|
|
|
|
if close_on_trigger:
|
|
body["closeOnTrigger"] = True
|
|
|
|
try:
|
|
data = bybit_v5_request("POST", "/v5/order/create", body=body)
|
|
return data.get("result", {})
|
|
except Exception as e:
|
|
raise ValueError(f"Failed to place order: {str(e)}")
|
|
|
|
def place_spot_order_with_sl_tp(
|
|
symbol: str,
|
|
side: str,
|
|
qty: float,
|
|
price: Optional[float] = None,
|
|
stop_loss_price: Optional[float] = None,
|
|
take_profit_price: Optional[float] = None,
|
|
sl_limit_price: Optional[float] = None,
|
|
tp_limit_price: Optional[float] = None,
|
|
sl_order_type: str = "Market",
|
|
tp_order_type: str = "Market",
|
|
order_type: str = "Limit",
|
|
time_in_force: str = "PostOnly"
|
|
) -> Dict:
|
|
"""
|
|
Convenience function to place a spot order with stop loss and take profit.
|
|
|
|
Args:
|
|
symbol: Trading pair symbol (e.g., "BTCUSDT")
|
|
side: Order side ("Buy" or "Sell")
|
|
qty: Order quantity
|
|
price: Limit price (None for market orders)
|
|
stop_loss_price: Stop loss trigger price
|
|
take_profit_price: Take profit trigger price
|
|
sl_limit_price: Stop loss limit price (for limit SL orders)
|
|
tp_limit_price: Take profit limit price (for limit TP orders)
|
|
sl_order_type: Stop loss order type ("Market" or "Limit")
|
|
tp_order_type: Take profit order type ("Market" or "Limit")
|
|
order_type: "Limit" or "Market"
|
|
time_in_force: Time in force ("GTC", "IOC", "FOK", "PostOnly")
|
|
|
|
Returns:
|
|
Dict containing order result
|
|
"""
|
|
return place_order(
|
|
symbol=symbol,
|
|
side=side,
|
|
order_type=order_type,
|
|
qty=qty,
|
|
price=price,
|
|
stop_loss=stop_loss_price,
|
|
take_profit=take_profit_price,
|
|
sl_limit_price=sl_limit_price,
|
|
tp_limit_price=tp_limit_price,
|
|
sl_order_type=sl_order_type,
|
|
tp_order_type=tp_order_type,
|
|
time_in_force=time_in_force,
|
|
category="spot"
|
|
)
|