代码块
import requests
import datetime
import math
import pandas as pd
from ta.momentum import RSIIndicator
import urllib3
import warnings
import logging
logging.basicConfig(level=logging.ERROR)
def get_btc_price():
"""
从币安API获取比特币当前价格
返回:
float: 比特币当前价格（以USDT计）
"""
try:
url = "https://api.binance.com/api/v3/ticker/price"
params = {"symbol": "BTCUSDT"}
response = requests.get(url, params=params)
data = response.json()
return float(data["price"])
except Exception as e:
logging.error(f"获取比特币价格时发生错误: {str(e)}")
return None
def get_btc_200ma():
"""
从币安API获取过去200天的比特币价格并计算平均值
这个函数获取过去200天的每日收盘价，并计算它们的平均值，
也就是200日定投成本。
返回:
float: 比特币200日平均价格
"""
try:
url = "https://api.binance.com/api/v3/klines"
params = {
"symbol": "BTCUSDT",
"interval": "1d",
"limit": 200
}
response = requests.get(url, params=params)
data = response.json()
prices = [float(item[4]) for item in data] # 收盘价是第5个元素（索引4）
return sum(prices) / len(prices)
except Exception as e:
logging.error(f"获取比特币200日平均价格时发生错误: {str(e)}")
return None
def calculate_bitcoin_age():
"""
计算比特币的年龄（以天为单位）
从比特币创世区块的日期（2009年1月3日）到今天的天数
返回:
int: 比特币年龄（天数）
"""
genesis_date = datetime.date(2009, 1, 3)
current_date = datetime.date.today()
return (current_date - genesis_date).days
def calculate_exponential_growth_valuation(bitcoin_age):
"""
计算比特币的指数增长估值
使用公式: 10^(5.84 * log10(比特币年龄) - 17.01)
参数:
bitcoin_age (int): 比特币年龄（天数）
返回:
float: 指数增长估值
"""
return 10 ** (5.84 * math.log10(bitcoin_age) - 17.01)
def calculate_ahr999(price, ma200, exp_growth_val):
"""
计算AHR999指标
使用公式: (比特币价格/200日定投成本) * (比特币价格/指数增长估值)
参数:
price (float): 当前比特币价格
ma200 (float): 200日平均价格（200日定投成本）
exp_growth_val (float): 指数增长估值
返回:
float: AHR999指标值
"""
return (price / ma200) * (price / exp_growth_val)
# def get_altcoin_season_index():
# """
# 获取代币季指数
# 代币季指数用于衡量山寨币相对比特币的表现。
# 指数范围从0到100：
# 0-25: 比特币季节
# 26-75: 中性市场
# 76-100: 山寨币季节
# 返回:
# int: 代币季指数值
# str: 市场状态描述
# """
# url = "https://api.blockchaincenter.net/v1/index/altcoin_season_index"
# response = requests.get(url, proxies=proxies)
# data = response.json()
# index = data["altcoin_season_index"]
# if index <= 25:
# status = "比特币季节"
# elif index <= 75:
# status = "中性市场"
# else:
# status = "山寨币季节"
# return index, status
def get_btc_price_on_date(date):
"""
从币安API获取指定日期的比特币收盘价
参数:
date (datetime.date): 指定的日期
返回:
float: 指定日期的比特币收盘价（USDT计）
"""
url = "https://api.binance.com/api/v3/klines"
# date转换为datetime，然后获取timestamp
datetime_obj = datetime.datetime.combine(date, datetime.time.min)
start_time = int(datetime_obj.timestamp() * 1000)
params = {
"symbol": "BTCUSDT",
"interval": "1d",
"startTime": start_time,
"limit": 1
}
response = requests.get(url, params=params)
data = response.json()
if data:
return float(data[0][4]) # 收盘价是第5个元素（索引4）
else:
raise ValueError(f"无法获取 {date} 的比特币价格数据")
def get_btc_200ma_on_date(date):
"""
从币安API获取指定日期之前200天的比特币价格并计算平均值
参数:
date (datetime.date): 指定的日期
返回:
float: 比特币200日平均价格
"""
# 将date转换为datetime，然后获取timestamp
datetime_obj = datetime.datetime.combine(date, datetime.time.min)
end_time = int(datetime_obj.timestamp() * 1000)
url = "https://api.binance.com/api/v3/klines"
params = {
"symbol": "BTCUSDT",
"interval": "1d",
"endTime": end_time,
"limit": 200
}
response = requests.get(url, params=params)
data = response.json()
prices = [float(item[4]) for item in data] # 收盘价是第5个元素（索引4）
return sum(prices) / len(prices)
def get_binance_c2c_rate():
"""
获取币安C2C的CNY对USDT汇率
返回:
float: CNY对USDT的汇率
"""
try:
url = "https://p2p.binance.com/bapi/c2c/v2/friendly/c2c/adv/search"
headers = {
"Content-Type": "application/json"
}
payload = {
"fiat": "CNY",
"page": 1,
"rows": 10,
"tradeType": "BUY",
"asset": "USDT",
"countries": [],
"proMerchantAds": False,
"publisherType": None,
"payTypes": []
}
response = requests.post(url, json=payload, headers=headers)
data = response.json()
if data["success"] and data["data"]:
rates = [float(ad["adv"]["price"]) for ad in data["data"]]
return sum(rates) / len(rates)
else:
raise ValueError("无法获取币安C2C的CNY对USDT汇率")
except Exception as e:
logging.error(f"获取币安C2C汇率时发生错误: {str(e)}")
return None
def get_fear_and_greed_index():
"""
获取恐惧与贪婪指数
- 0-24 ：极度恐惧（橙色）
- 25-49: 恐惧(琥珀色/黄色）
- 50-74: 贪婪(浅绿色）
- 75 -100: 极度贪婪(绿色）
"""
try:
url = "https://api.alternative.me/fng/"
response = requests.get(url)
data = response.json()
return int(data["data"][0]["value"])
except Exception as e:
logging.error(f"获取恐惧与贪婪指数时发生错误: {str(e)}")
return None
# def get_google_trends():
# """
# 获取谷歌趋势数据
# """
# pytrends = TrendReq(hl='en-US', tz=360)
# kw_list = ["bitcoin"]
# pytrends.build_payload(kw_list, timeframe='today 1-m')
# data = pytrends.interest_over_time()
# return data["bitcoin"].iloc[-1]
def get_week_200ma():
"""
获取200周移动平均线
"""
url = "https://api.binance.com/api/v3/klines"
params = {
"symbol": "BTCUSDT",
"interval": "1w",
"limit": 200
}
response = requests.get(url, params=params)
data = response.json()
prices = [float(item[4]) for item in data] # 收盘价是第5个元素（索引4）
return sum(prices) / len(prices)
def get_week_rsi():
"""
获取周RSI
"""
url = "https://api.binance.com/api/v3/klines"
params = {
"symbol": "BTCUSDT",
"interval": "1w",
"limit": 100
}
response = requests.get(url, params=params)
data = response.json()
prices = [float(item[4]) for item in data] # 收盘价是第5个元素（索引4）
df = pd.DataFrame(prices, columns=['close'])
rsi_indicator = RSIIndicator(close=df['close'], window=14)
return rsi_indicator.rsi().iloc[-1]
# def get_market_bias():
# """
# 计算市场偏差（Market Bias）
# 这里我们使用一个简单的方法：比较当前价格与200周均线的关系
# """
# current_price = get_btc_price()
# ma200w = get_week_200ma()
# bias = (current_price - ma200w) / ma200w * 100
# return bias
def get_binance_btc_data(symbol="BTCUSDT", interval="1h", limit=100):
"""
从币安API获取比特币K线数据
"""
url = "https://api.binance.com/api/v3/klines"
params = {
"symbol": symbol,
"interval": interval,
"limit": limit
}
response = requests.get(url, params=params)
data = response.json()
df = pd.DataFrame(data, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume', 'close_time', 'quote_asset_volume', 'number_of_trades', 'taker_buy_base_asset_volume', 'taker_buy_quote_asset_volume', 'ignore'])
df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms')
df['close'] = df['close'].astype(float)
df['volume'] = df['volume'].astype(float)
return df
def calculate_can_short(price_data, volume_data, ma_period=20, volume_threshold=1.5, price_change_threshold=0.02):
"""
根据价格和交易量数据计算是否可以做空
"""
# 计算价格移动平均线
price_ma = price_data.rolling(window=ma_period).mean()
# 计算交易量移动平均线
volume_ma = volume_data.rolling(window=ma_period).mean()
# 检查当前价格是否低于移动平均线
price_below_ma = price_data.iloc[-1] < price_ma.iloc[-1]
# 检查当前交易量是否高于平均水平
volume_above_average = volume_data.iloc[-1] > volume_ma.iloc[-1] * volume_threshold
# 计算价格变化
price_change = (price_data.iloc[-1] - price_data.iloc[-2]) / price_data.iloc[-2]
# 检查价格是否显著下跌
significant_price_decrease = price_change < -price_change_threshold
# 判断是否可以做空
can_short = 1 if price_below_ma and volume_above_average and significant_price_decrease else 0
return can_short
def can_short():
"""
判断是否可以做空
"""
try:
# 从币安获取比特币数据
df = get_binance_btc_data()
# 计算can_short值
can_short_value = calculate_can_short(df['close'], df['volume'])
return can_short_value
except Exception as e:
logging.error(f"计算can_short时发生错误: {str(e)}")
return None
def main(date=None):
"""
主函数，协调整个计算过程并输出结果
参数:
date (datetime.date, 可选): 指定的日期，默认为None（使用当前日期）
"""
try:
if date is None:
date = datetime.date.today()
# 获取指定日期的比特币价格
btc_price = get_btc_price_on_date(date)
# 计算指定日期的200日平均价格
btc_200ma = get_btc_200ma_on_date(date)
# 计算比特币年龄
bitcoin_age = (date - datetime.date(2009, 1, 3)).days
# 计算指数增长估值
exp_growth_val = calculate_exponential_growth_valuation(bitcoin_age)
# 计算AHR999指标，处理可能的None值
ahr999 = None
if btc_price and btc_200ma and exp_growth_val:
ahr999 = calculate_ahr999(btc_price, btc_200ma, exp_growth_val)
# 获取币安C2C的CNY对USDT汇率
cny_usdt_rate = get_binance_c2c_rate()
# 获取新增的指标
fgi = get_fear_and_greed_index()
# google_trends = get_google_trends()
week_200ma = get_week_200ma()
week_rsi = get_week_rsi()
# market_bias = get_market_bias()
can_short_value = can_short()
# 输出结果，处理可能的None值
print(f"日期: {date}")
print(f"比特币价格: ${btc_price:.2f}" if btc_price else "比特币价格: 无法获取")
print(f"比特币200日平均价格: ${btc_200ma:.2f}" if btc_200ma else "比特币200日平均价格: 无法获取")
print(f"比特币年龄（天）: {bitcoin_age}")
print(f"指数增长估值: ${exp_growth_val:.2f}" if exp_growth_val else "指数增长估值: 无法计算")
print(f"AHR999 指标值: {ahr999:.4f}" if ahr999 else "AHR999 指标值: 无法计算")
print(f"币安C2C CNY/USDT 汇率: ¥{cny_usdt_rate:.2f}" if cny_usdt_rate else "币安C2C CNY/USDT 汇率: 无法获取")
if btc_price and cny_usdt_rate:
print(f"比特币价格（CNY）: ¥{btc_price * cny_usdt_rate:.2f}")
else:
print("比特币价格（CNY）: 无法计算")
print(f"恐惧与贪婪指数: {fgi}" if fgi is not None else "恐惧与贪婪指数: 无法获取")
# print(f"谷歌趋势数据: {google_trends}" if google_trends is not None else "谷歌趋势数据: 无法获取")
print(f"200周移动平均线: ${week_200ma:.2f}" if week_200ma else "200周移动平均线: 无法获取")
print(f"周RSI: {week_rsi:.2f}" if week_rsi is not None else "周RSI: 无法获取")
# print(f"市场偏差: {market_bias:.2f}%" if market_bias is not None else "市场偏差: 无法计算")
print(f"是否可以做空: {'是' if can_short_value == 1 else '否'}" if can_short_value is not None else "是否可以做空: 无法判断")
except Exception as e:
logging.error(f"主函数执行时发生错误: {str(e)}")
print("程序执行过程中发生错误，请查看日志获取详细信息。")
if __name__ == "__main__":
main()
）
