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

部分代码更新

Browse Source
This commit is contained in:
fengche
2025-10-11 18:06:48 +08:00
parent c69702248a
commit 5febbcb0ab
4 changed files with 1041 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

191
coinbus/nochain_eth_lyq.py Normal file
View File

@@ -0,0 +1,191 @@
# coding=utf-8
import ujson
from binance.websocket.spot.websocket_client import SpotWebsocketClient as WebsocketClient
import time
import requests
#from loguru import logger
import datetime
import pymysql
import math
class NochainDbIf:
def __init__(self, host="172.17.0.1", port=4423, user="root", password="2GS@bPYcgiMyL14A", dbname="ethdb"):
self.conn = pymysql.connect(host=host, port=port, user=user, password=password, database=dbname,
cursorclass=pymysql.cursors.DictCursor)
print("init nochain db suceess!")
def save(self, day, price, ma350x2, ma111, ma350x1, ma350x1r6, ma350x3, ma350x5):
with self.conn.cursor() as cursor:
sql_insert = "REPLACE INTO `nochainv3a` (`unixdt`, `price`, `ma350x2`, `ma111`, ma350x1, ma350x1r6, ma350x3, ma350x5) VALUES (FROM_UNIXTIME(%s), %s, %s, %s, %s, %s, %s, %s)"
# print(sql_insert)
cursor.execute(sql_insert, (day, price, ma350x2, ma111, ma350x1, ma350x1r6, ma350x3, ma350x5))
self.conn.commit()
def prepare_maxxx(prices, day, madays):
total = 0
cnt = 0
for i in range(madays):
if day in prices:
total += prices[day]
cnt += 1
# print(day, total, cnt)
day = str(int(day) - 3600 * 24)
if cnt > 0:
return total / cnt
return 0
def prepare_ma350(prices, day):
return prepare_maxxx(prices, day, 350)
def prepare_ma111(prices, day):
return prepare_maxxx(prices, day, 111)
def prepare_gold_ratio(prices):
ma350x1 = {}
ma350x1r6 = {}
ma350x2 = {}
ma350x3 = {}
ma350x5 = {}
for day in prices:
ma350x1[day] = prepare_maxxx(prices, day, 350)
ma350x1r6[day] = ma350x1[day] * 1.6
ma350x2[day] = ma350x1[day] * 2
ma350x3[day] = ma350x1[day] * 3
ma350x5[day] = ma350x1[day] * 5
return ma350x1, ma350x1r6, ma350x2, ma350x3, ma350x5
def calc_pi_cycle_top(dbif, prices):
ma350x2 = {}
ma111 = {}
for day in prices:
ma350x2[day] = prepare_ma350(prices, day) * 2
ma111[day] = prepare_ma111(prices, day)
return ma350x2, ma111
def get_current_utc():
curtime = time.gmtime(time.time())
daystr = time.strftime("%d %b %Y", curtime)
dayutc = int(time.mktime(time.strptime(daystr, "%d %b %Y")))
return dayutc
def get_current_price():
url = "https://data.messari.io/api/v1/assets/eth/metrics/market-data&interval=1d"
header_set = {}
header_set["x-messari-api-key"] = "aH2pyj5i4QGo1k1gLxXEbIJ5RJr+FYKLEWk6cRT6RuSc6lRY"
response_price = requests.get(url, headers=header_set)
if response_price.status_code == 200:
priceweb = ujson.loads(response_price.content)
if "data" in priceweb:
priceset = priceweb["data"]
if "market_data" in priceset:
pricedata = priceset["market_data"]
if "price_usd" in pricedata:
price = pricedata["price_usd"]
return price
return None
def get_history_price(coin_id):
prices = {}
dayutc = get_current_utc()
price = get_current_price()
if price is not None:
prices[str(dayutc)] = price
print("start...", dayutc, price)
dayt = time.gmtime()
daystr = time.strftime("%Y", dayt)
year = int(daystr)
end_year = year
while True:
# if end_year < 2022:
# break
url = ""
if end_year != year:
start_year = end_year
url = "https://data.messari.io/api/v1/assets/" + coin_id + "/metrics/price/time-series?start="
else:
url = "https://data.messari.io/api/v1/assets/" + coin_id + "/metrics/price/time-series?after=" + str(
year) + "-01-01&order=descending&interval=1d"
# now_time = time.gmtime()
# daystr = time.strftime("%Y-%m-%d", now_time)
# url = url + daystr + "&order=desc&format=json"
if end_year != year:
url = url + str(start_year) + "-01-01&end=" + str(end_year) + "-12-31&interval=1d&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_supply = requests.get(url, headers=header_set)
# print(response_supply)
if response_supply.status_code == 200:
#print(response_supply.content)
supplyweb = ujson.loads(response_supply.content)
if "data" in supplyweb:
supplyset = supplyweb["data"]
if "values" in supplyset:
valueset = supplyset["values"]
if valueset is not None:
for supply in valueset:
dayutc = int(supply[0] / 1000)
s = supply[1]
prices[str(dayutc)] = float(s)
# print(s, dayutc, supplys[str(dayutc)])
# break
else:
break
else:
break
end_year -= 1
time.sleep(2)
return prices
def get_eth_history_price():
return get_history_price("ethereum")
def nochain():
global dbif
dbif = NochainDbIf()
print("prepare...")
prices = get_eth_history_price()
#print(prices)
ma350x2, ma111 = calc_pi_cycle_top(dbif, prices)
print("calc_pi_cycle_top ok.")
ma350x1, ma350x1r6, ma350x2, ma350x3, ma350x5 = prepare_gold_ratio(prices);
print("prepare_gold_ratio ok.")
for day in prices:
#print(day)
ma350x21 = 0
if day in ma350x2:
ma350x21 = ma350x2[day]
ma1111 = 0
if day in ma111:
ma1111 = ma111[day]
ma350x11 = 0
if day in ma350x1:
ma350x11 = ma350x1[day]
ma350x1r61 = 0
if day in ma350x1r6:
ma350x1r61 = ma350x1r6[day]
ma350x31 = 0
if day in ma350x3:
ma350x31 = ma350x3[day]
ma350x51 = 0
if day in ma350x5:
ma350x51 = ma350x5[day]
# print(day, prices[day], ma350x21, ma1111, supply, issue, s2f_ratio1, s2f_deflection1)
dbif.save(int(day), prices[day], ma350x21, ma1111, ma350x11,
ma350x1r61, ma350x31, ma350x51)
#print("save ok.")
nochain()

27
coinbus/nochain_lyq_utc08.py Normal file
View File

@@ -0,0 +1,27 @@
import subprocess
from datetime import datetime, timedelta
import time
while True:
# 获取当前时间的UTC时间
now = datetime.utcnow()
# 计算到下一个08:00的时间间隔
next_run = datetime(now.year, now.month, now.day, 8, 0)
if now >= next_run:
next_run += timedelta(days=1)
sleep_time = (next_run - now).total_seconds()
# 休眠直到下一个08:00
time.sleep(sleep_time)
# 运行 nochain_lyq_v2.py
command1 = f"python3 nochain_lyq_v2.py"
subprocess.run(command1, shell=True)
# 运行 nochain_update_lyq.py
command2 = f"python3 nochain_update_lyq.py"
subprocess.run(command2, shell=True)
# 运行 nochain_eth_lyq.py
command3 = f"python3 nochain_eth_lyq.py"
subprocess.run(command3, shell=True)

736
coinbus/nochain_lyq_v2.py Normal file
View File

@@ -0,0 +1,736 @@
# coding=utf-8
import ujson
from binance.websocket.spot.websocket_client import SpotWebsocketClient as WebsocketClient
import time
import requests
#from loguru import logger
from datetime import datetime
import pymysql
import math
from datetime import datetime
from stock_indicators import indicators
from stock_indicators.indicators.common.quote import Quote
class NochainDbIf:
def __init__(self, host="172.17.0.1", port=4423, user="root", password="2GS@bPYcgiMyL14A", dbname="btcdb"):
self.conn = pymysql.connect(host=host, port=port, user=user, password=password, database=dbname, cursorclass=pymysql.cursors.DictCursor)
print("init nochain db suceess!")
def save(self, day, price, ma350x2, ma111, supply, flow, s2fratio, s2fdeflection, ma350x1, ma350x1r6, ma350x3, ma350x5):
with self.conn.cursor() as cursor:
sql_insert = "REPLACE INTO `nochainv3c` (`unixdt`, `price`, `ma350x2`, `ma111`, `btcsupply`, `flow`, `s2fratio`, `s2fdeflection`, ma350x1, ma350x1r6, ma350x3, ma350x5) VALUES (FROM_UNIXTIME(%s), %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)"
#print(sql_insert)
cursor.execute(sql_insert, (day, price, ma350x2, ma111, supply, flow, s2fratio, s2fdeflection, ma350x1, ma350x1r6, ma350x3, ma350x5))
self.conn.commit()
def save_ssr(self, day, price, marketcap, usdtsupply, usdcsupply, busdsupply, daisupply, stables_supply, ssr, ssrosc):
with self.conn.cursor() as cursor:
sql_insert = "REPLACE INTO `nochainv3b` (`unixdt`, `price`, marketcap, usdtsupply, usdcsupply, busdsupply, daisupply, stables_supply, ssr, ssrosc) VALUES (FROM_UNIXTIME(%s), %s, %s, %s, %s, %s, %s, %s, %s, %s)"
#print(sql_insert)
cursor.execute(sql_insert, (day, price, marketcap, usdtsupply, usdcsupply, busdsupply, daisupply, stables_supply, ssr, ssrosc))
self.conn.commit()
# 看起来您已经定义了一个名为 NochainDbIf Python 类，该类与 MySQL 数据库交互。此类具有将数据保存到两个不同表（nochainv2c 和nochainv2b ）的方法。save该方法用于保存与加密货币价格和指标相关的数据，而save_ssr该方法用于保存与稳定币供应和SSR（稳定币供应比率）相关的数据。
# 下面是每种方法的作用的细分：
# __init__：这是使用提供的参数（主机、端口、用户、密码和数据库名称）初始化数据库连接的构造函数方法。它使用pymysql库来建立连接。
# save：此方法在nochainv2c表中插入或更新记录。它采用代表与加密货币价格和供应相关的各种指标的参数。
# save_ssr：此方法在nochainv2b表中插入或更新记录。它采用代表与稳定币供应和 SSR 相关的指标的参数。
# 这两种方法都使用 SQL 语句，REPLACE INTO该语句尝试在表中插入新行。如果存在主键或唯一键冲突的行，则改为更新现有行。
# 此外，这些方法使用上下文管理器 with self.conn.cursor() as cursor 来确保数据库资源得到正确管理，并在执行 SQL 语句后提交事务 self.conn.commit() 以保留更改。
def get_history_price():
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), prices[str(int(dayutc))])
return prices
# 此函数似乎可以从纳斯达克 API 检索历史加密货币价格。其工作原理如下：get_history_price
# 它初始化一个空字典，用于存储历史价格。prices
# 它使用该函数向纳斯达克 API 端点发送 GET 请求。requests.get()
# 如果响应状态代码为 200（表示成功），则继续分析响应数据。
# 它使用模块中的 JSON 响应加载到 Python 字典中。ujson.loads()ujson
# 它检查响应字典中是否存在该键，如果存在，则检索与之关联的密钥。"dataset""data"
# 它遍历每个价格数据点，提取日期和价格。
# 它将日期字符串转换为 Unix 时间戳，然后使用 解析 。time.mktime() time.strptime()
# 它将时间戳作为字符串键存储在字典中，并将相应的价格作为值。prices
# 最后，它返回包含历史价格的字典。prices
def get_history_price2(pricedict):
#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?start=" + str(
year) + "-01-01&end="+str(year)+"-12-31&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)
return pricedict
# 此函数似乎从 Messari API 获取历史加密货币价格并将其存储在字典中。让我们分解一下它是如何工作的：get_history_price2
# 它需要一个字典pricedict作为参数，大概是为了存储历史价格。
# 它初始化一些与时间相关的变量，包括当前年份 year 和结束年份end_year，后者最初设置为当前年份。
# 它进入一个循环，无限期地持续下去，直到明确中断。
# 在循环中，它构造 URL，用于根据start_year 和end_year 值从 Messari API 获取历史价格数据。
# 它设置一个包含用于身份验证的 Messari API 密钥的标头。
# 它使用构造的 URL 和标头向 Messari API 发送 GET 请求。requests.get()
# 如果响应状态代码为 200（表示成功），则会解析 JSON 响应。
# 它从响应中提取价格数据并对其进行迭代。
# 对于每个价格数据点，它将时间戳从毫秒转换为秒，将其转换为人类可读的日期格式，并将其作为 Unix 时间戳字符串键存储在字典pricedict中，并以相应的价格作为值。
# 它会递减变量以获取上一年的数据，并在下一次迭代前增加 2 秒的延迟。end_year
# 循环一直持续到获取所有年份的历史数据或发生错误为止。
# 最后，它打印包含获取的历史价格的字典pricedict并返回它。
def get_history_supply(supplys):
#supplys = {}
#issues = {}
response_supply = requests.get(
'https://data.nasdaq.com/api/v3/datatables/QDL/BCHAIN?code=TOTBC&api_key=FZqXog4sR-b7cYnXcRVV')
if response_supply.status_code == 200:
#print(response_supply.content)
supplyweb = ujson.loads(response_supply.content)
if "datatable" in supplyweb:
supplyset = supplyweb["datatable"]
# if "end_date" in supplyset:
# end_date = supplyset["end_date"]
if "data" in supplyset:
supplydata = supplyset["data"]
#previssue = 0
#prevday = 0
for supply in supplydata:
daystr = supply[1]
p = supply[2]
dayutc = time.mktime(time.strptime(daystr, "%Y-%m-%d"))
supplys[str(int(dayutc))] = float(p)
#if supply[0] == end_date:
#previssue = float(p)
#prevday = dayutc
#else:
#issues[str(int(prevday))] = previssue - supplys[str(int(dayutc))]
#previssue = float(p)
#prevday = dayutc
#print(supply, int(dayutc), supply[str(int(dayutc))])
#return supplys, issues
print("get_history_supply", supplys)
return supplys
# 此功能似乎从纳斯达克 API 检索与比特币总供应量相关的历史数据。以下是其工作原理的细分：get_history_supply
# 它采用一个字典supplys作为参数，大概是为了存储历史供应数据。
# 它使用该函数向纳斯达克 API 端点发送 GET 请求。requests.get()
# 如果响应状态代码为 200（表示成功），则继续分析响应数据。
# 它使用模块中的 JSON 响应加载到 Python 字典中。ujson.loads() ujson
# 它检查响应字典中是否存在该键，如果存在，则检索与之关联的密钥。"dataset""data"
# 它提取数据集的结束日期，大概是为了进行一些进一步的处理（似乎被注释掉了）。
# 它遍历数据集中的每个供应数据点。
# 对于每个数据点，它提取日期和比特币总供应量，在用 解析后将日期字符串转换为 Unix 时间戳，并将时间戳作为字符串键存储在字典supplys中，并以相应的供应值作为值。time.mktime() time.strptime()
# 最后，它打印包含获取的历史供应数据的字典supplys并返回它。
# 似乎有一些注释掉的部分与计算连续天数之间的供应差异并将该信息存储在另一个名为issues 的字典中有关，但这些部分目前处于非活动状态。
def get_history_supply2():
supplys = {}
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/sply-circ/time-series?start="
else:
url = "https://data.messari.io/api/v1/assets/bitcoin/metrics/sply-circ/time-series?start=" + str(
year) + "-01-01&end="+str(year)+"-12-31&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_csupply = requests.get(url, headers=header_set)
# print(response_price)
if response_csupply.status_code == 200:
# print(response_price.content)
csweb = ujson.loads(response_csupply.content)
if "data" in csweb:
csset = csweb["data"]
if "values" in csset:
valueset = csset["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")))
supplys[str(ret_dayutc)] = float(s)
#print(s, dayutc, supplys[str(dayutc)])
#break
else:
break
else:
break
break
end_year -= 1
time.sleep(2)
return supplys
# 此函数似乎从 Messari API 获取历史比特币供应数据并将其存储在字典中。以下是其工作原理的细分：get_history_supply2
# 它初始化一个空字典supplys，用于存储历史供应数据。
# 它获取当前年份并将变量设置为当前年份。end_year
# 它进入一个循环，无限期地持续下去，直到明确中断。
# 在循环中，它构造 URL，用于根据 start_year和 end_year值从 Messari API 获取历史供应数据。
# 它设置一个包含用于身份验证的 Messari API 密钥的标头。
# 它使用构造的 URL 和标头向 Messari API 发送 GET 请求。requests.get()
# 如果响应状态代码为 200（表示成功），则会解析 JSON 响应。
# 它从响应中提取供应数据并对其进行迭代。
# 对于每个供应数据点，它将时间戳从毫秒转换为秒，将其转换为人类可读的日期格式，并将其作为 Unix 时间戳字符串键存储在字典supplys中，并以相应的供应值作为值。
# 它打印包含获取的历史供应数据的字典supplys并返回它。
# 但是，在循环的第一次迭代之后有一个break语句，该语句在仅获取数据一年后有效地终止了循环。这可能是无意的。
def calc_issues(supplys):
issues = {}
prevday = 0
previssue = 0
for supply_day in supplys:
if int(supply_day) > prevday:
prevday = int(supply_day)
previssue = supplys[supply_day]
print("calc_issues", prevday, previssue, supplys)
for dayutc in supplys:
issues[str(int(prevday))] = previssue - supplys[str(int(dayutc))]
previssue = float(supplys[dayutc])
prevday = dayutc
print(issues)
return issues
# 此函数似乎根据提供的历史供应数据计算连续几天之间的比特币供应量差异。让我们分解一下它是如何工作的：calc_issues
# 它需要一个称为supplys输入的字典，其中可能包含历史比特币供应数据。
# 它初始化一个空字典，该字典issues用于存储计算出的供应差异。
# 它初始化变量，并分别跟踪前一天的时间戳和供应量。prevday previssue
# 它会遍历字典supplys中每天的供应数据。
# 如果当天的时间戳大于前一天的时间戳，则它会更新为当天的时间戳和供应量。prevday previssue
# 它计算前一天和当天之间的供应量差异，并将其存储在字典issues中，并将前一天的时间戳作为键。
# 它会更新并用于下一次迭代。previssue prevday
# 它打印出计算出的供应差异。
# 它返回包含计算出的供应差异的字典issues。
# 但是，实现存在潜在问题。prevday和 previssue变量用于跟踪前一天的供应量，但循环会覆盖每个新日期的供应量。因此，字典issues将仅包含字典supplys中最后一天与所有其他日期之间的差异。
def prepare_year_issues(issues):
issues_year = {}
temp_issues_year = {}
for day in issues:
dayt = time.gmtime(int(day))
daystr = time.strftime("%Y", dayt)
if daystr in temp_issues_year:
temp_issues_year[daystr] = temp_issues_year[daystr] + issues[day]
else:
temp_issues_year[daystr] = issues[day]
#print(temp_issues_year)
for day in issues:
dayt = time.gmtime(int(day))
daystr = time.strftime("%Y", dayt)
if daystr in temp_issues_year:
issues_year[day] = temp_issues_year[daystr]
#print(issues_year)
return issues_year
# “prepare_year_issues”功能似乎将每日发行数据汇总为年度总数。其工作原理如下：
# - 它使用一个名为“issues”的字典作为输入，其中可能包含每日发行数据。
# - 它初始化了两个字典：“issues_year”用于存储汇总的年度发行数据，“temp_issues_year”用于临时存储每年的发行总和。
# - 它每天都在“issues”字典中迭代。
# - 对于每一天，它都会从当天的 Unix 时间戳中提取年份。
# - 它检查年份是否已经存在于“temp_issues_year”中。
# - 如果是，则将该日的发行量添加到相应年份的总数中。
# - 如果不是，它会以“temp_issues_year”的形式初始化该日期的年份。
# - 然后，它每天再次在“issues”字典中迭代。
# - 对于每一天，它都会提取年份并检查年份是否以“temp_issues_year”形式存在。
# - 如果是，则在“issues_year”字典中为该日期分配相应的年度总发行量。
# 最后，它返回包含年度汇总发行数据的“issues_year”字典。
# 此功能有效地将每日发行数据汇总为年度总数，从而更容易进行分析和可视化。
def prepare_maxxx(prices, day, madays):
total = 0
cnt = 0
for i in range(madays):
if day in prices:
total += prices[day]
cnt += 1
# print(day, total, cnt)
day = str(int(day) - 3600 * 24)
if cnt > 0:
return total / cnt
return 0
# “prepare_maxxx”函数似乎计算了特定日期（“天”）结束的指定天数（“madays”）内加密货币价格的移动平均线。以下是其工作原理的细分：
# - 它需要三个参数：
# - 'prices'：包含以 Unix 时间戳为键的加密货币价格的字典。
# - 'day'：一个 Unix 时间戳，表示计算移动平均线的周期的结束日期。
# - 'madays'：一个整数，表示计算移动平均线的天数。
# - 它初始化“总计”和“cnt”变量，以分别计算移动平均线中包含的价格总和和价格计数。
# - 它进入一个循环，迭代“madays”时间，表示计算移动平均线的天数。
# - 在每次迭代中：
# - 它检查当前“日”是否存在于“价格”字典中。
# - 如果是这样，它会将当天的价格添加到“总计”中，并增加“cnt”。
# - 然后，它将“一天”减少 24 小时（3600 秒）以移动到前一天。
# - 循环后，它会检查计算中是否包含任何价格（如果“cnt”大于 0）。
# - 如果是这样，它将返回指定时间段内的平均价格（“总计/cnt”）。
# - 如果不是（即，如果在指定时间段内没有找到价格），则返回 0。
# 此函数计算在特定日期结束的指定天数内加密货币价格的简单移动平均线。
def prepare_ma350(prices, day):
return prepare_maxxx(prices, day, 350)
# 用于计算加密货币价格的 350 天移动平均线。它通过调用“prepare_maxxx”函数并将“madays”参数设置为 350 来实现这一点。以下是其工作原理的摘要：
# - 它需要两个参数：
# - 'prices'：包含以 Unix 时间戳为键的加密货币价格的字典。
# - 'day'：一个 Unix 时间戳，表示计算移动平均线的周期的结束日期。
# - 它只是返回调用“prepare_maxxx（prices， day， 350）”的结果。
# 此函数抽象了计算移动平均线的细节，并为获取加密货币价格的 350 天移动平均线提供了更简洁的界面。
def prepare_ma111(prices, day):
return prepare_maxxx(prices, day, 111)
# “prepare_ma111”函数类似于“prepare_ma350”函数，但计算加密货币价格的 111 天移动平均线。其工作原理如下：
# - 它需要两个参数：
# - 'prices'：包含以 Unix 时间戳为键的加密货币价格的字典。
# - 'day'：一个 Unix 时间戳，表示计算移动平均线的周期的结束日期。
# - 它返回调用“prepare_maxxx（prices， day， 111）”的结果。
# 此函数提供了一种方便的方法来计算加密货币价格的 111 天移动平均线，而无需每次明确指定天数。
def prepare_ma200(ssr, day):
return prepare_maxxx(ssr, day, 200)
# “prepare_ma200”函数类似于之前的移动平均线函数，但它计算的是 200 天移动平均线。让我们分解一下它的功能：
# -**参数**：
# - 'ssr'：包含与稳定币供应比率 （SSR） 相关的数据的字典，以 Unix 时间戳为键。
# - 'day'：一个 Unix 时间戳，表示计算移动平均线的周期的结束日期。
# -**功能性**：
# - 它调用“prepare_maxxx”函数，并将“madays”参数设置为 200，传递“ssr”字典和指定的“day”。
# - 此函数抽象了计算 200 天移动平均线的细节，并为获取它提供了更简洁的界面。
# 此函数用于计算与稳定币供应比率 （SSR） 相关的数据的 200 天移动平均线
def prepare_gold_ratio(prices):
ma350x1 = {}
ma350x1r6 = {}
ma350x2 = {}
ma350x3 = {}
ma350x5 = {}
for day in prices:
ma350x1[day] = prepare_maxxx(prices, day, 350)
ma350x1r6[day] = ma350x1[day]*1.6
ma350x2[day] = ma350x1[day] * 2
ma350x3[day] = ma350x1[day] * 3
ma350x5[day] = ma350x1[day] * 5
return ma350x1, ma350x1r6, ma350x2, ma350x3, ma350x5
# “prepare_gold_ratio”函数似乎根据一组给定的价格计算各种移动平均线。其工作原理如下：
# -**参数**：
# - 'prices'：包含以 Unix 时间戳为键的加密货币价格的字典。
# -**功能性**：
# - 它初始化字典以存储不同的移动平均线：'ma350x1'、'ma350x1r6'、'ma350x2'、'ma350x3' 和 'ma350x5'。
# - 它在“价格”字典中每天都会迭代。
# - 对于每一天，它使用“prepare_maxxx”函数计算 350 天移动平均线并将其存储在“ma350x1”中。
# - 它将“ma350x1r6”计算为“ma350x1”值的 1.6 倍。
# - 它将“ma350x2”、“ma350x3”和“ma350x5”分别计算为“ma350x1”值的 2、3 和 5 倍。
# - 在所有天数遍历后，它返回计算出的移动平均线：'ma350x1'、'ma350x1r6'、'ma350x2'、'ma350x3' 和 'ma350x5'。
# 此功能旨在根据加密货币价格的 350 天移动平均线准备不同的比率，可能用于比较或分析目的
def cal_ssr_osc(ssr):
ssr_osc = {}
for day in ssr:
ssr_ma = prepare_ma200(ssr, day)
ssr_osc[day] = ssr[day]/ssr_ma
return ssr_osc
# “cal_ssr_osc”功能根据提供的 SSR 数据计算稳定币供应比率 （SSR） 的振荡器。其工作原理如下：
# -**参数**：
# - 'ssr'：包含与稳定币供应比率 （SSR） 相关的数据的字典，以 Unix 时间戳为键。
# -**功能性**：
# - 它初始化一个名为“ssr_osc”的字典来存储计算出的 SSR 振荡器值。
# - 它在“ssr”字典中每天迭代。
# - 对于每一天，它使用“prepare_ma200”函数计算 SSR 的 200 天移动平均线。
# - 它将当天的 SSR 值除以 200 天移动平均线，以计算 SSR 振荡器值。
# - 它将 SSR 振荡器值存储在“ssr_osc”字典中，并以相应的日期作为键。
# - 在所有天数遍历后，它会返回包含计算出的 SSR 振荡器值的“ssr_osc”字典。
# 此函数计算 SSR 与其 200 天移动平均线的比率，代表 SSR 振荡器，可用于分析与稳定币供应动态相关的趋势或信号
def calc_pi_cycle_top(dbif, prices):
ma350x2 = {}
ma111 = {}
for day in prices:
ma350x2[day] = prepare_ma350(prices, day)*2
ma111[day] = prepare_ma111(prices, day)
return ma350x2, ma111
# “calc_pi_cycle_top”函数根据提供的价格计算两条移动平均线，“ma350x2”和“ma111”。以下是其功能的细分：
# -**参数**：
# - 'dbif'：此参数似乎未在函数中使用，因此可能用于将来使用。
# - 'prices'：包含以 Unix 时间戳为键的加密货币价格的字典。
# -**功能性**：
# - 它初始化字典“ma350x2”和“ma111”以存储计算出的移动平均线。
# - 它在“价格”字典中每天都会迭代。
# - 对于每一天，它计算：
# - 'ma350x2'：使用“prepare_ma350”功能是 350 天移动平均线价格的两倍。
# - 'ma111'：使用“prepare_ma111”函数的 111 天移动平均价格。
# - 在所有日期迭代后，它返回计算出的移动平均线“ma350x2”和“ma111”。
# 此功能似乎是通过分析特定移动平均线来识别加密货币价格周期中潜在峰值的过程的一部分
def calc_s2f_ratio(dbif, prices, supplys, issues_year):
s2f_ratio = {}
s2f_deflection = {}
cnt = 1
for day in supplys:
if day in issues_year:
s2f = 0
if int(day) >= 1672502400: #2023
s2f = supplys[day]/(900*365)
else:
s2f = supplys[day]/issues_year[day]
print(s2f,day,supplys[day],issues_year[day])
s2f_ratio[day] = 0.09*(math.pow(s2f, 3.3))
#print(supplys[day], issues_year[day], s2f, s2f_ratio[day])
s2f_deflection[day] = prices[day]/s2f_ratio[day]
#print(day, prices[day], s2f, s2f_ratio[day], s2f_deflection[day])
#cnt+=1
#if cnt > 10:
#break
return s2f_ratio, s2f_deflection
# “calc_s2f_ratio”功能根据提供的数据计算与比特币的库存流量 （S2F） 模型相关的两个指标。让我们分解一下它的功能：
# -**参数**：
# - 'dbif'：此参数似乎未在函数中使用，因此可能用于将来使用。
# - 'prices'：包含以 Unix 时间戳为键的加密货币价格的字典。
# - 'supplys'：包含与加密货币供应相关的数据的字典，以 Unix 时间戳为键。
# - “issues_year”：包含年度汇总发行数据的字典。
# -**功能性**：
# - 它初始化两个字典，“s2f_ratio”和“s2f_deflection”，分别存储计算出的 S2F 比率和 S2F 偏转指标。
# - 它初始化一个值为 1 的计数器 'cnt'（似乎未使用）。
# - 它在“供应”字典中每天迭代。
# - 对于每一天，它都会检查该日期是否存在于“issues_year”字典中，确保该日有相应的发行数据可用。
# - 它根据可用的供应和发行数据计算库存流量 （S2F） 比率。如果该日期在 2023 年 1 月 1 日之后（以 Unix 时间戳1672502400表示），则使用固定发行率（每天 900 BTC）计算 S2F 比率。否则，它使用供应量除以年度发行量来计算 S2F 比率。
# - 它使用涉及幂和乘法的公式计算 S2F 比率值。
# - 它通过将当天的价格除以当天的 S2F 比率来计算 S2F 偏转。
# - 它将计算出的 S2F 比率和 S2F 挠度值存储在各自的字典中，并以相应的日期为键。
# - 在遍历所有天数后，它会返回包含计算指标的“s2f_ratio”和“s2f_deflection”字典。
# 这个功能似乎是分析比特币的库存流量模型的关键部分，提供了对其与价格动态和潜在偏差的关系的见解。
def get_stable_coin_supply(coin_id):
supplys = {}
dayt = time.gmtime()
daystr = time.strftime("%Y", dayt)
year = int(daystr)
end_year = year
#split_2023 = False
while True:
#if end_year < 2022:
#break
url = ""
#if split_2023 and end_year != year:
if end_year != year:
start_year = end_year
url = "https://data.messari.io/api/v1/assets/"+coin_id+"/metrics/sply-circ/time-series?start="
else:
#if split_2023:
# url = "https://data.messari.io/api/v1/assets/" + coin_id + "/metrics/sply-circ/time-series?after=" + str(
# year) + "-06-01&order=descending"
#else:
# url = "https://data.messari.io/api/v1/assets/" + coin_id + "/metrics/sply-circ/time-series?start=2023-01-01&end=2023-06-01&order=descending"
url = "https://data.messari.io/api/v1/assets/" + coin_id + "/metrics/sply-circ/time-series?start=" + str(
year) + "-01-01&end="+str(year)+"-12-31&order=descending&interval=1d"
#now_time = time.gmtime()
#daystr = time.strftime("%Y-%m-%d", now_time)
#url = url + daystr + "&order=desc&format=json"
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_supply = requests.get(url, headers=header_set)
#print(response_supply)
if response_supply.status_code == 200:
#print(response_supply.content)
supplyweb = ujson.loads(response_supply.content)
if "data" in supplyweb:
supplyset = supplyweb["data"]
if "values" in supplyset:
valueset = supplyset["values"]
if valueset is not None:
for supply in valueset:
dayutc = int(supply[0]/1000)
s = supply[1]
supplys[str(dayutc)] = float(s)
#print(s, dayutc, supplys[str(dayutc)])
#break
else:
break
else:
break
#if split_2023:
end_year -= 1
#else:
# split_2023 = True
time.sleep(2)
return supplys
# “get_stable_coin_supply”功能根据给定稳定币的代币 ID 检索其历史供应数据。让我们来看看它是如何工作的：
# -**论点**：
# - 'coin_id'：要获取历史供应数据的稳定币的标识符。
# -**功能性**：
# - 它初始化一个名为“supplys”的空字典来存储检索到的供应数据。
# - 它检索当前年份并将“end_year”设置为当前年份。
# - 它进入一个循环，一直持续到数据检索完成。
# - 在循环中，它构造了 URL，用于根据 'coin_id' 和 'end_year' 从 Messari API 获取历史供应数据。
# - 它使用“requests.get（）”向 Messari API 发送 GET 请求，其中包含包含 Messari API 密钥的构造 URL 和标头。
# - 如果响应状态码为 200（表示成功），则解析 JSON 响应。
# - 它从响应中提取供应数据并对其进行迭代。
# - 对于每个供应数据点，它将时间戳从毫秒转换为秒，并将其作为 Unix 时间戳字符串键存储在“供应”字典中，并以相应的供应值作为值。
# - 循环继续，直到检索到所有可用的历史供应数据。
# - 它将“end_year”递减 1，以便在下一次迭代中获取上一年的数据。
# - 在API请求之间等待2秒，避免服务器过载。
# - 检索到所有历史供应数据后，它将返回包含数据的“供应”字典。
# 此功能可以检索给定稳定币的历史供应数据，便于分析和监控其随时间推移的供应动态
def get_usdt_supply():
return get_stable_coin_supply("tether")
# “get_usdt_supply
# 它直接调用“get_stable_coin_supply（“tether”）
# 它返回从 'get_stable_coin_supply（“tether”） 获得的结果
# 该功能提供了一种方便的方式来获取 USDT 供应数据，而无需每次都指定代币 ID
def get_usdc_supply():
return get_stable_coin_supply("usd-coin")
# “get_usdc_sup但专门设计用于检索 USD Coin （USDC） 的历史供应数据。其工作原理如下：
# 它直接调用“get_stable_coin_supply("usd-coin")
# 它返回从“get_stable_coin_supply（“binance-usd")中获取的结果
# 此功能提供了一种方便的方式来获取 USDC 供应数据，而无需每次都指定硬币 ID
def get_busd_supply():
return get_stable_coin_supply("binance-usd")
# “get_busd_suppget_stable_coin_supply函数，这次专门用于检索 Binance USD （BUSD） 的历史供应数据。其工作原理如下：
# 它直接调用“get_stable_coin_supply("binance-usd"),该参数是 Binance USD （BUSD） 的硬币 ID
# 它返回从 'get_stable_coin_supply（“binance-usd")中获取的结果
# 此函数提供了一种方便的方式来获取 BUSD 供应数据，而无需在每个时间指定硬币 ID
def get_dai_supply():
return get_stable_coin_supply("dai")
# 该函数是该函数的另一个包装器，专门用于检索 Dai （DAI） 的历史供应数据。以下是它的运作方式：get_dai_supply get_stable_coin_supply
# 它直接调用get_stable_coin_supply"dai"函数，传递参数，该参数对应于 Dai （DAI） 的硬币 ID。
# 此函数返回从get_stable_coin_supply("dai")中获取的结果。
# 此功能简化了获取 DAI 供应数据的过程，无需每次都指定硬币 ID
def cal_ssr(dbif, prices, supplys):
print("calc_ssr")
usdts = get_usdt_supply()
#print(usdts)
#return
usdcs = get_usdc_supply()
#print(usdcs)
busds = get_busd_supply()
dais = get_dai_supply()
print("calc_ssr start")
stables = {}
ssr = {}
marketcap = {}
for day in usdts:
stables[day] = usdts[day]
if day in usdcs:
stables[day] += usdcs[day]
if day in busds:
stables[day] += busds[day]
if day in dais:
stables[day] += dais[day]
#print(day, stables[day], usdts[day], usdcs[day], busds[day], dais[day])
#print(prices[day])
if day in prices:
#print(day, prices)
if day in supplys:
#print(day, supplys)
marketcap[day] = prices[day] * supplys[day]
if stables[day] > 0:
ssr[day] = marketcap[day]/stables[day]
else:
ssr[day] = 0
#break
print(ssr)
ssrosc = {}
quotes_list = []
for day in ssr:
#print(day)
dayt = time.gmtime(int(day))
#dayt = datetime.date.fromtimestamp(int(day))
daystr = time.strftime("%Y-%m-%d", dayt)
dtobj = datetime.strptime(daystr, "%Y-%m-%d")
print(dtobj)
quotes_list.append(Quote(dtobj, 0, 0, 0, ssr[day], 0))
print(quotes_list)
ssr_osc = indicators.get_bollinger_bands(quotes_list, 200, 2)
for item in ssr_osc:
if item.z_score is not None:
ssrosc[str(int(item.date.timestamp()))] = item.z_score
#ssrosc = cal_ssr_osc(ssr)
print(ssrosc)
for day in ssr:
price = 0
if day in prices:
price = prices[day]
marketcap1 = 0
if day in marketcap:
marketcap1 = marketcap[day]
usdt = 0
if day in usdts:
usdt = usdts[day]
usdc = 0
if day in usdcs:
usdc = usdcs[day]
busd = 0
if day in busds:
busd = busds[day]
dai = 0
if day in dais:
dai = dais[day]
stable = 0
if day in stables:
stable = stables[day]
ssr1 = 0
if day in ssr:
ssr1 = ssr[day]
ssrosc1 = 0
if day in ssrosc:
ssrosc1 = ssrosc[day]
dbif.save_ssr(day, price, marketcap1, usdt, usdc, busd, dai,
stable, ssr1, ssrosc1)
# 似乎“cal_ssr”功能旨在计算稳定币供应比率 （SSR） 及其相应的振荡器 （SSR Oscillator） 并将其保存到数据库中。以下是该函数的运行方式：
# -**参数**：
# - 'dbif'：数据库接口类的一个实例，尽管它在当前实现中似乎未使用。
# - 'prices'：包含以 Unix 时间戳为键的加密货币价格的字典。
# - 'supplys'：包含以 Unix 时间戳为键的稳定币历史供应数据的字典。
# -**功能性**：
# - 它分别使用专用函数“get_usdt_supply”、“get_usdc_supply”、“get_busd_supply”和“get_dai_supply”获取不同稳定币的历史供应数据，包括 USDT、USDC、BUSD 和 DAI。
# - 它初始化字典以存储稳定币供应量、SSR、市值和 SSR 振荡器数据。
# - 它遍历所提供数据中的可用日期。
# - 对于每个日期，它通过将 USDT、USDC、BUSD 和 DAI 的供应量相加来计算稳定币的总供应量。
# - 它根据加密货币的价格及其相应的供应量计算市值。
# - 它通过将市值除以稳定币总供应量来计算 SSR。
# - 它构造一个包含每个日期的 SSR 值的“Quote”对象列表。
# - 它使用周期为 200 天且标准差为 2 的布林带计算 SSR 振荡器。
# - 它使用数据库接口类的“save_ssr”方法将 SSR 和 SSR 振荡器值保存到数据库中。
# 总体而言，该功能似乎提供了对稳定币供应比率及其相关指标的全面分析，有助于进一步研究和监控稳定币动态
def nochain():
global dbif
dbif = NochainDbIf()
temp_prices = get_history_price()
prices = get_history_price2(temp_prices)
#supplys, issues = get_history_supply()
temp_supplys = get_history_supply2()
supplys = get_history_supply(temp_supplys)
issues = calc_issues(supplys)
'''
cnt = 0;
for day in supplys:
print(supplys[day])
if day in issues:
print(issues[day])
cnt+=1
if cnt > 5:
break
'''
issues_year = prepare_year_issues(issues)
#print(issues_year)
ma350x2, ma111 = calc_pi_cycle_top(dbif, prices)
#print(ma350x2)
#print(ma111)
s2f_ratio, s2f_deflection = calc_s2f_ratio(dbif, prices, supplys, issues_year)
#print(s2f_ratio)
#print(s2f_deflection)
ma350x1, ma350x1r6, ma350x2, ma350x3, ma350x5 = prepare_gold_ratio(prices)
for day in prices:
#print(day)
ma350x21 = 0
if day in ma350x2:
ma350x21 = ma350x2[day]
ma1111 = 0
if day in ma111:
ma1111 = ma111[day]
supply = 0
if day in supplys:
supply = supplys[day]
issue = 0
if day in issues:
issue = issues[day]
s2f_ratio1 = 0
if day in s2f_ratio:
s2f_ratio1 = s2f_ratio[day]
s2f_deflection1 = 0
if day in s2f_deflection:
s2f_deflection1 = s2f_deflection[day]
ma350x11 = 0
if day in ma350x1:
ma350x11 = ma350x1[day]
ma350x1r61 = 0
if day in ma350x1r6:
ma350x1r61 = ma350x1r6[day]
ma350x31 = 0
if day in ma350x3:
ma350x31 = ma350x3[day]
ma350x51 = 0
if day in ma350x5:
ma350x51 = ma350x5[day]
#print(day, prices[day], ma350x21, ma1111, supply, issue, s2f_ratio1, s2f_deflection1)
dbif.save(int(day), prices[day], ma350x21, ma1111, supply, issue, s2f_ratio1, s2f_deflection1, ma350x11, ma350x1r61, ma350x31, ma350x51)
cal_ssr(dbif, prices, supplys)
# “nochain”功能负责协调NoChain项目的数据收集、处理和存储。以下是它的运作方式：
# -**初始化**：
# - 初始化 'NochainDbIf' 类的全局实例 'dbif'，表示数据库接口。
# - **数据检索**：
# - 它使用“get_history_price”检索历史价格数据。
# - 它使用“get_history_supply2”检索历史供应数据。
# - 它使用“calc_issues”根据历史供应数据计算问题。
# - 它使用“prepare_year_issues”准备年度汇总问题。
# - 它使用“calc_pi_cycle_top”计算 Pi Cycle 顶部指标（“ma350x2”和“ma111”）。
# - 它使用“calc_s2f_ratio”计算库存流量 （S2F） 比率和挠度。
# - 它使用“prepare_gold_ratio”准备黄金比率指标（'ma350x1'、'ma350x1r6'、'ma350x2'、'ma350x3'、'ma350x5'）。
# - **数据存储**：
# - 对于价格数据中的每一天：
# - 它检索相关指标。
# - 它使用“NochainDbIf”的“save”方法将数据存储到数据库中。
# - **稳定币供应比率（SSR）计算**：
# - 它使用“cal_ssr”计算 SSR 及其振荡器，利用来自“get_usdt_supply”、“get_usdc_supply”、“get_busd_supply”和“get_dai_supply”的数据。
# 总体而言，该功能是收集、处理和存储与NoChain项目相关的数据的主要驱动力，从而实现后续分析和可视化。
nochain()

87
coinbus/nochain_update_lyq.py Normal file
View File

@@ -0,0 +1,87 @@
# coding=utf-8
import ujson
from binance.websocket.spot.websocket_client import SpotWebsocketClient as WebsocketClient
import time
import requests
import datetime
import pymysql
import math
from stock_indicators import indicators
from stock_indicators.indicators.common.quote import Quote
class NochainDbIf:
def __init__(self, host="172.17.0.1", port=4423, user="root", password="2GS@bPYcgiMyL14A", dbname="btcdb"):
self.conn = pymysql.connect(host=host, port=port, user=user, password=password, database=dbname, cursorclass=pymysql.cursors.DictCursor)
print("init nochain db suceess!")
# 可能用于与数据库交互。该方法是构造函数，它使用库设置与MySQL数据库的连接。
# 以下是该方法的作用的细分：__init__
# 它需要几个参数这些参数用于连接到MySQL数据库。如果未提供这些参数，则使用默认值。
# 在该方法中，它使用提供的参数建立与MySQL数据库的连接。
# 该参数指定结果应作为字典而不是元组返回，这样可以更方便地访问数据。cursorclass = pymysql.cursors.DictCursor
# 最后，它打印一条消息，指示数据库连接的初始化成功。
def get_ssr(self):
with self.conn.cursor() as cursor:
sql_cmd = "SELECT unixdt,ssr FROM nochainv3b order by unixdt"
cursor.execute(sql_cmd)
self.conn.commit()
return cursor.fetchall()
# 该方法似乎旨在从名为nochainv2b的MySQL数据库表中检索数据。下面是此方法的细分：get_ssr
# 它使用上下文管理器 （） 自动处理游标，并确保在执行查询后正确关闭连接。with
# 在上下文管理器中，它准备一个SQL命令来从nochainv2b表中选择数据。该命令选择unixdt和ssr列，并按unixdt对结果进行排序。
# 它使用光标的方法执行SQL命令。execute
# 执行命令后，它使用.如果对数据库进行了任何需要永久保存的更改，则此步骤是必需的。self.conn.commit()
# 最后，它返回查询的结果 ，它以字典列表的形式检索查询返回的所有行，其中每个字典表示一行，其中列名作为键和相应的值。cursor.fetchall()
def update_ssr(self, dt, ssrosc):
with self.conn.cursor() as cursor:
sql_update = "UPDATE nochainv3b SET unixdt='%s', ssrosc='%s' WHERE unixdt='%s'" %(dt, ssrosc, dt)
print(sql_update)
cursor.execute(sql_update)
self.conn.commit()
# 该方法将更新MySQL数据库nochainv2b表中的数据。下面是此方法的细分：update_ssr
# 它需要两个参数：dt（可能表示时间戳）和ssrosc（可能表示与SSR相关的某个值）。
# 在该方法中，它准备一个SQLUPDATE命令来更新nochainv2b表中特定列的列。
# 该字符串的格式为dt和ssrosc的值，以动态构造SQL命令。这是潜在的风险，因为它开启了SQL注入攻击的可能性。改用参数化查询更安全。
# 然后，该方法打印出构造的SQL命令以进行调试。
# 它使用光标的方法执行SQL命令。execute
# 执行命令后，它会提交事务，使更改永久化。self.conn.commit()
# 但是，SQL命令中存在一个潜在问题：在设置新值和WHERE子句的条件时使用相同的值。这可能不按预期运行，因为它将更新与给定.
# 如果打算更新特定行，则可能需要相应地调整WHERE子句。
def rollback(self):
self.conn.rollback()
# 它对当前数据库连接执行回滚操作。以下是它的作用：rollback
# 它在数据库连接 （） 上调用该方法。rollback self.conn
# 此操作将当前事务期间所做的任何未提交的更改还原到以前的状态。
# 当事务期间出现错误时，通常使用回滚，允许您还原在错误发生之前所做的任何更改。
def nochain():
try:
dbif = NochainDbIf()
ssr_ma = dbif.get_ssr()
#print(ssr_ma)
quotes_list = []
for item in ssr_ma:
print(item["unixdt"])
quotes_list.append(Quote(item["unixdt"],0,0,0,item["ssr"],0))
#print(quotes_list)
ssr_osc = indicators.get_bollinger_bands(quotes_list, 200, 2)
for item in ssr_osc:
if item.z_score is not None:
#print(item.date, item.sma, item.percent_b, item.z_score, item.width)
#dbif.update_ssr(item.date, item.z_score)
break
print("ok")
except Exception as e:
#dbif.rollback();
print(e)
finally:
print("end")
nochain()
# 可能使用该nochain类计算从 MySQL 数据库获取的 SSR（大概是卖空比率）数据上的布林带。NochainDbIf
# 以下是该函数功能的细分：
# 它首先创建类的实例以建立与数据库的连接。NochainDbIf
# 它使用实例的方法从数据库中检索 SSR 数据。get_ssr NochainDbIf
# 它使用检索到的 SSR 数据构造对象列表。Quote
# 它使用模块中的函数计算布林带。get_bollinger_bandsindicators
# 它遍历计算出的布林带，并打印有关每个带的一些信息。
# 目前，它似乎已注释掉了使用计算出的布林带更新数据库中 SSR 数据的部分。
# 如果在此过程中发生任何异常，它将捕获异常，打印它，然后继续。
# 最后，无论该过程是成功还是遇到错误，它都会打印“结束”