lizixuan/coinbus-data
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

utxos代码更新

Browse Source
This commit is contained in:
fengche
2025-10-17 16:15:17 +08:00
parent 5febbcb0ab
commit e1848ce48b
2 changed files with 329 additions and 96 deletions
Download Patch File Download Diff File Expand all files Collapse all files

125
coinbus/btc_update.py Normal file
View File

@@ -0,0 +1,125 @@
import pymysql
import json
import os
import time
from datetime import datetime, timedelta
# 数据库配置
DB_CONFIG = {
"host": "192.168.194.240",
"user": "root",
"password": "2GS@bPYcgiMyL14A",
"database": "btcdb",
"port": 4423,
"connect_timeout": 60,
"read_timeout": 60,
"write_timeout": 60,
"charset": "utf8mb4"
}
# 数据文件路径
DATA_FILE = "btc_historical_price.py"
# 定时任务间隔（秒）—— 例如 3600 为每小时更新一次
INTERVAL = 28800
def get_new_prices(source, last_timestamp=None):
"""
从数据库获取 source 数据源的最新价格
仅每天北京时间 08:00:00 的数据减 8 小时存入文件
"""
conn = pymysql.connect(**DB_CONFIG)
prices = {}
try:
with conn.cursor() as cursor:
if last_timestamp:
sql = """
SELECT timestamp, price
FROM btc_prices
WHERE source = %s AND timestamp > %s
ORDER BY timestamp
"""
cursor.execute(sql, (source, last_timestamp))
else:
sql = """
SELECT timestamp, price
FROM btc_prices
WHERE source = %s
ORDER BY timestamp
"""
cursor.execute(sql, (source,))
rows = cursor.fetchall()
for timestamp, price in rows:
ts_int = int(timestamp)
# 转换为北京时间
dt_beijing = datetime.utcfromtimestamp(ts_int) + timedelta(hours=8)
# 如果是每天 08:00:00 北京时间，则减 8 小时
if dt_beijing.hour == 8 and dt_beijing.minute == 0 and dt_beijing.second == 0:
ts_int -= 8 * 3600
prices[str(ts_int)] = float(price)
finally:
conn.close()
return prices
def load_existing_data():
"""加载历史价格数据"""
if not os.path.exists(DATA_FILE):
return {}, {}
try:
with open(DATA_FILE, "r", encoding="utf-8") as f:
ns = {}
exec(f.read(), ns)
return ns.get("prices_temp", {}), ns.get("prices", {})
except Exception:
return {}, {}
def save_prices(prices_temp, prices):
"""保存价格数据到文件"""
with open(DATA_FILE, "w", encoding="utf-8") as f:
f.write("# 自动生成的BTC历史价格数据文件\n")
f.write(f"# 更新时间: {datetime.now()}\n\n")
f.write("prices_temp = ")
f.write(json.dumps(prices_temp, indent=4, ensure_ascii=False))
f.write("\n\nprices = ")
f.write(json.dumps(prices, indent=4, ensure_ascii=False))
f.write("\n")
def get_last_timestamp(price_dict):
"""获取当前字典中最大的时间戳"""
if not price_dict:
return None
return max(int(ts) for ts in price_dict.keys())
def update_once():
"""执行一次更新流程"""
prices_temp, prices = load_existing_data()
last_nasdaq_ts = get_last_timestamp(prices_temp)
last_crypto_ts = get_last_timestamp(prices)
nasdaq_new = get_new_prices("Nasdaq", last_nasdaq_ts)
crypto_new = get_new_prices("CryptoCompare", last_crypto_ts)
prices_temp.update(nasdaq_new)
prices.update(crypto_new)
save_prices(prices_temp, prices)
def main():
"""主循环任务"""
while True:
try:
update_once()
except Exception:
pass
time.sleep(INTERVAL)
if __name__ == "__main__":
main()

300
coinbus/btc_utxos_lyq2.py
View File

@@ -203,9 +203,10 @@ class UtxosIf:
addr = None
addrtype = None
ip = "127.0.0.1"
port = " 8332"
port = "8332"
user = "user"
password = "password"
timeout=100
if rpc is None:
rpc = RPC(ip, port, user, password)
tx = None
@@ -1169,24 +1170,59 @@ class UtxosIf:
# 对于MongoDB中的每个集合，该方法将数据插入到集合中，并打印插入的文档以进行验证。
# 此外，该方法还会计算和更新一些其他指标（如利润率、实现价格、短期和长期持有者的利润率、相对比率等），并将它们插入“utxosv2”表中。
# 总体而言，这种方法有助于将分析的数据存储到关系数据库 （MySQL） 和NoSQL数据库 （MongoDB） 中，从而便于检索和进一步分析。
def get_history_price(self):
prices = {}
response_price = requests.get(
'https://data.nasdaq.com/api/v3/datatables/QDL/BCHAIN?code=MKPRU;api_key=FZqXog4sR-b7cYnXcRVV')
if response_price.status_code == 200:
#print(response_price.content)
priceweb = ujson.loads(response_price.content)
if "datatable" in priceweb:
priceset = priceweb["datatable"]
if "data" in priceset:
pricedata = priceset["data"]
for price in pricedata:
daystr = price[1]
p = price[2]
dayutc = time.mktime(time.strptime(daystr, "%Y-%m-%d"))
prices[str(int(dayutc))] = float(p)
#print(price, int(dayutc), g_prices[str(int(dayutc))])
return prices
def get_history_price(self, batch_size=5000):
"""获取数据库中的 Nasdaq 数据，存入字典"""
db_config = {
"host": "192.168.194.240",
"user": "root",
"password": "2GS@bPYcgiMyL14A",
"database": "btcdb",
"port": 4423,
"connect_timeout": 60,
"read_timeout": 60,
"write_timeout": 60,
"charset": "utf8mb4"
}
offset = 0
self.pricedict = {}
while True:
connection = pymysql.connect(**db_config)
try:
with connection.cursor() as cursor:
sql = "SELECT timestamp, price FROM btc_prices WHERE source = 'Nasdaq' ORDER BY timestamp LIMIT %s OFFSET %s"
cursor.execute(sql, (batch_size, offset))
rows = cursor.fetchall()
if not rows:
break
for timestamp, price in rows:
self.pricedict[str(int(timestamp))] = float(price)
finally:
connection.close()
offset += batch_size
if len(rows) < batch_size:
break # 最后一页读取完成
return self.pricedict
#prices = {}
#response_price = requests.get(
# 'https://data.nasdaq.com/api/v3/datatables/QDL/BCHAIN?code=MKPRU;api_key=FZqXog4sR-b7cYnXcRVV')
#if response_price.status_code == 200:
# #print(response_price.content)
# priceweb = ujson.loads(response_price.content)
# if "datatable" in priceweb:
# priceset = priceweb["datatable"]
# if "data" in priceset:
# pricedata = priceset["data"]
# for price in pricedata:
# daystr = price[1]
# p = price[2]
# dayutc = time.mktime(time.strptime(daystr, "%Y-%m-%d"))
# prices[str(int(dayutc))] = float(p)
# #print(price, int(dayutc), g_prices[str(int(dayutc))])
#return prices
# “get_history_price”方法似乎用于从WebAPI终结点检索历史价格数据。以下是它的作用的细分：
# - 初始化一个空字典“prices”来存储历史价格数据。
# - 向指定的API端点发送HTTP GET请求，该端点可能提供历史比特币价格数据。
@@ -1200,54 +1236,95 @@ class UtxosIf:
# - 使用'strptime'将日期字符串 （'daystr'） 解析为Unix时间戳 （'dayutc'）。
# - 将Unix 时间戳作为键存储在“prices”字典中，并将相应的price （'p'） 作为值。
# 最后，它返回包含历史价格数据的“prices”字典，其中Unix时间戳作为键，价格作为值
def get_history_price2(self, pricedict):
def get_history_price2(self, batch_size=5000):
#pricedict = {}
dayt = time.gmtime()
daystr = time.strftime("%Y", dayt)
year = int(daystr)
end_year = year
while True:
url = ""
if end_year != year:
start_year = end_year
url = "https://data.messari.io/api/v1/assets/bitcoin/metrics/price/time-series?start="
else:
url = "https://data.messari.io/api/v1/assets/bitcoin/metrics/price/time-series?after=" + str(
year) + "-01-01&order=descending&interval=1d"
"""获取数据库中的 Messari 数据，存入字典"""
db_config = {
"host": "192.168.194.240",
"user": "root",
"password": "2GS@bPYcgiMyL14A",
"database": "btcdb",
"port": 4423,
"connect_timeout": 60,
"read_timeout": 60,
"write_timeout": 60,
"charset": "utf8mb4"
}
if end_year != year:
url = url + str(start_year) + "-01-01&end=" + str(end_year) + "-12-31&order=descending&interval=1d"
header_set = {}
header_set["x-messari-api-key"] = "aH2pyj5i4QGo1k1gLxXEbIJ5RJr+FYKLEWk6cRT6RuSc6lRY"
# header_set["Content-Type"] = "application/json"
print(header_set, url)
response_price = requests.get(url, headers=header_set)
# print(response_price)
if response_price.status_code == 200:
# print(response_price.content)
priceweb = ujson.loads(response_price.content)
if "data" in priceweb:
priceset = priceweb["data"]
if "values" in priceset:
valueset = priceset["values"]
if valueset is not None:
for supply in valueset:
dayutc = int(supply[0] / 1000)
s = supply[1]
ret_time = time.gmtime(dayutc)
ret_daystr = time.strftime("%d %b %Y", ret_time)
ret_dayutc = int(time.mktime(time.strptime(ret_daystr, "%d %b %Y")))
pricedict[str(ret_dayutc)] = float(s)
# print(s, dayutc, pricedict[str(dayutc)])
# break
else:
break
else:
break
end_year -= 1
time.sleep(2)
print(pricedict)
return pricedict
offset = 0
self.pricedict = {}
while True:
connection = pymysql.connect(**db_config)
try:
with connection.cursor() as cursor:
sql = """
SELECT timestamp, price
FROM btc_prices
WHERE source = 'CryptoCompare'
ORDER BY timestamp
LIMIT %s OFFSET %s
"""
cursor.execute(sql, (batch_size, offset))
rows = cursor.fetchall()
if not rows:
break
for timestamp, price in rows:
self.pricedict[str(int(timestamp))] = float(price)
finally:
connection.close()
offset += batch_size
if len(rows) < batch_size:
break # 数据已全部读取
return self.pricedict
#dayt = time.gmtime()
#daystr = time.strftime("%Y", dayt)
#year = int(daystr)
#end_year = year
#while True:
# url = ""
# if end_year != year:
# start_year = end_year
# url = "https://data.messari.io/api/v1/assets/bitcoin/metrics/price/time-series?start="
# else:
# url = "https://data.messari.io/api/v1/assets/bitcoin/metrics/price/time-series?after=" + str(
# year) + "-01-01&order=descending&interval=1d"
#
# if end_year != year:
# url = url + str(start_year) + "-01-01&end=" + str(end_year) + "-12-31&order=descending&interval=1d"
# header_set = {}
# header_set["x-messari-api-key"] = "aH2pyj5i4QGo1k1gLxXEbIJ5RJr+FYKLEWk6cRT6RuSc6lRY"
# # header_set["Content-Type"] = "application/json"
# print(header_set, url)
# response_price = requests.get(url, headers=header_set)
# # print(response_price)
# if response_price.status_code == 200:
# # print(response_price.content)
# priceweb = ujson.loads(response_price.content)
# if "data" in priceweb:
# priceset = priceweb["data"]
# if "values" in priceset:
# valueset = priceset["values"]
# if valueset is not None:
# for supply in valueset:
# dayutc = int(supply[0] / 1000)
# s = supply[1]
# ret_time = time.gmtime(dayutc)
# ret_daystr = time.strftime("%d %b %Y", ret_time)
# ret_dayutc = int(time.mktime(time.strptime(ret_daystr, "%d %b %Y")))
# pricedict[str(ret_dayutc)] = float(s)
# # print(s, dayutc, pricedict[str(dayutc)])
# # break
# else:
# break
# else:
# break
# end_year -= 1
# time.sleep(2)
# print(pricedict)
# return pricedict
# 'get_history
# 初始化与时间相关的变量，包括当前年份。
# 进入一个循环，从当前年份开始，向后循环访问年份。
@@ -1264,21 +1341,46 @@ class UtxosIf:
# 最后，它返回“pricedict”
def get_current_price(self):
price = 0
DB_CONFIG = {
"host": "192.168.194.240",
"user": "root",
"password": "2GS@bPYcgiMyL14A",
"database": "btcdb",
"port": 4423
}
connection = pymysql.connect(**DB_CONFIG)
try:
response_price = requests.get(
'https://api.binance.com/api/v3/ticker/price?symbol=BTCUSDT')
prices = ujson.loads(response_price.text)
price = float(prices["price"])
print(response_price.text, price)
response_price.close()
# print("price", price)
return price
except:
response_price = requests.get("https://api.coinpaprika.com/v1/tickers/btc-bitcoin")
prices = ujson.loads(response_price.text)
price = float(prices["quotes"]["USD"]["price"])
response_price.close()
return price
with connection.cursor() as cursor:
for source in ("binance", "coinbase"):
cursor.execute("""
SELECT price FROM btc_realtime_prices
WHERE source=%s
ORDER BY timestamp DESC
LIMIT 1
""", (source,))
row = cursor.fetchone()
if row:
price = float(row[0])
break
finally:
connection.close()
return price
#price = 0
#try:
# response_price = requests.get(
# 'https://api.binance.com/api/v3/ticker/price?symbol=BTCUSDT')
# prices = ujson.loads(response_price.text)
# price = float(prices["price"])
# print(response_price.text, price)
# response_price.close()
# # print("price", price)
# return price
#except:
# response_price = requests.get("https://api.coinpaprika.com/v1/tickers/btc-bitcoin")
# prices = ujson.loads(response_price.text)
# price = float(prices["quotes"]["USD"]["price"])
# response_price.close()
# return price
# 该函数似乎是一种从特定
# API端点检索比特币当前价格的方法。以下是其功能的细分：get_current_price
# 初始化默认值为0的变量。price
@@ -1393,14 +1495,15 @@ class UtxosIf:
password = "password"
rpc = RPC(ip, port, user, password)
# total_height = rpc.blockchain.get_block_count()
while True:
try:
total_height = rpc.blockchain.get_block_count()
break
except Exception as e:
print(f"❌ get_block_count 超时或异常：{e}，重试中...")
time.sleep(3)
rpc = RPC(ip, port, user, password) # 重新建立连接
print("rpctimeout")
time.sleep(10)
rpc = RPC(ip, port, user, password)
if height >= total_height:
return
prev_height = None
@@ -1488,14 +1591,15 @@ class UtxosIf:
password = "password"
rpc = RPC(ip, port, user, password)
# total_height = rpc.blockchain.get_block_count()
while True:
try:
total_height = rpc.blockchain.get_block_count()
break
except Exception as e:
print(f"❌ get_block_count 超时或异常：{e}，重试中...")
print("rpctimeout")
time.sleep(10)
rpc = RPC(ip, port, user, password) # 重新建立连接
rpc = RPC(ip, port, user, password)
if height >= total_height:
return
#total_height = rpc.blockchain.get_block_count()
@@ -1557,6 +1661,7 @@ class UtxosIf:
#current_price = self.get_current_price()
#if current_price == 0:
# return
print("handle_utxos initiate")
connin = sqlite3.connect("utxos.sqlite")
cin = connin.cursor()
cursorin = cin.execute("SELECT * from utxos")
@@ -1581,12 +1686,13 @@ class UtxosIf:
daystr = time.strftime("%d %b %Y", dt2)
dayutc = int(time.mktime(time.strptime(daystr, "%d %b %Y")))
price = 0
dayutc=dayutc-28800
current_price=current_price-28800
print("ceshi",str(dayutc))
print("ceshi2", str(current_price))
if str(dayutc) > str(check_dayutc):
continue
d_t = datetime.utcfromtimestamp(dayutc) + timedelta(hours=8)
d_t08am = datetime(d_t.year, d_t.month, d_t.day, 8, 0, 0)
dayutc=int((d_t08am - timedelta(hours=8)).timestamp())-28800
d_t2 = datetime.utcfromtimestamp(check_dayutc) + timedelta(hours=8)
d_t208am = datetime(d_t2.year, d_t2.month, d_t2.day, 8, 0, 0)
check_dayutc=int((d_t208am - timedelta(hours=8)).timestamp())-28800
print("ceshi",5)
if str(dayutc) in prices:
price = int(prices[str(dayutc)])
else:
@@ -1617,7 +1723,9 @@ class UtxosIf:
self.lth_mvrv = self.lth_mv / self.lth_rcap
if self.sth_rcap > 0:
self.sth_mvrv = self.sth_mv / self.sth_rcap
print("save_db initiate")
self.save_db()
print("save_db ok")
cin.close()
#cout.close()
@@ -1694,15 +1802,15 @@ if __name__ == '__main__':
os.system("if [ -e utxos.dat ]; then rm utxos.dat; fi")
os.system("if [ -e utxos.sqlite ]; then rm utxos.sqlite; fi")
check_dayutc = int(time.mktime(time.strptime(check_dt, "%Y-%m-%d")))
cmd = "~/bitcoin-29.0/bin/bitcoin-cli -rpcuser=user -rpcpassword=password invalidateblock " + daily_hash[str(check_dayutc)]
cmd = "~/bitcoin-29.0/bin/bitcoin-cli -rpcuser=user -rpcpassword=password -rpcclienttimeout=600 invalidateblock " + daily_hash[str(check_dayutc)]
os.system(cmd)
print("select ok")
print("select ok",daily_hash[str(check_dayutc)])
time.sleep(60);
os.system("~/bitcoin-29.0/bin/bitcoin-cli -rpcuser=user -rpcpassword=password dumptxoutset ~/utxos.dat")
os.system("~/bitcoin-29.0/bin/bitcoin-cli -rpcuser=user -rpcpassword=password -rpcclienttimeout=1800 dumptxoutset ~/utxos.dat latest")
print("dumptxoutset ok")
time.sleep(60);
# os.system("./utxo_to_sqlite ./utxos.dat ./utxos.sqlite")
os.system("python3 ~/utxo_to_sqlite.py ~/utxos.dat ~/utxos.sqlite")
#os.system("./utxo_to_sqlite ./utxos.dat ./utxos.sqlite")
os.system("python3 utxo_to_sqlite.py ./utxos.dat ./utxos.sqlite")
print("utxo_to_sqlite ok")
time.sleep(60);
stats.utxos(check_dayutc)
@@ -1727,4 +1835,4 @@ if __name__ == '__main__':
# - 它使用“utxos”方法处理指定日期的 UTXO。
# - 它使用比特币 CLI 命令“reconsiderblock”重新考虑之前失效的区块。
# 8. 每个步骤都附有指示脚本进度的打印语句。
# 这个脚本似乎是一个更大的系统的一部分，该系统与比特币网络交互并执行与UTXO相关的任务。它严重依赖比特币核心软件和命令行界面（“bitcoin-cli”）
# 这个脚本似乎是一个更大的系统的一部分，该系统与比特币网络交互并执行与UTXO相关的任务。它严重依赖比特币核心软件和命令行界面（“bitcoin-cli”）