import requests import csv from datetime import datetime import subprocess import time import logging # 配置常量 GIT_INTERVAL = 0 # Git提交间隔(秒) FETCH_INTERVAL = 3600 # 数据获取间隔(秒) LOG_FILE = "btc_plus.log" def timestamp_to_datetime(timestamp): """将时间戳转换为可读时间字符串""" return ( datetime.fromtimestamp(timestamp).isoformat() if timestamp else datetime.now().isoformat() ) def get_current_time(): """获取当前时间字符串""" return datetime.now().isoformat() def fetch_btc_plus_reward(address, stage_no=1): """从API获取BTC Plus奖励数据""" url = "https://graphql.sft-api.com/graphql" headers = { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:141.0) Gecko/20100101 Firefox/141.0", "Accept": "*/*", "Accept-Language": "zh-CN,zh;q=0.8,en-US;q=0.5,en;q=0.3", "Accept-Encoding": "gzip, deflate, br, zstd", "Referer": "https://app.solv.finance/", "content-type": "application/json", "authorization": "dW5kZWZpbmVkfHx1bmRlZmluZWR8fHVuZGVmaW5lZHx8.undefined", "x-amz-user-agent": "aws-amplify/3.0.7", "Origin": "https://app.solv.finance", "DNT": "1", "Connection": "keep-alive", "Sec-Fetch-Dest": "empty", "Sec-Fetch-Mode": "cors", "Sec-Fetch-Site": "cross-site", "Sec-GPC": "1", "Priority": "u=4", } query = """ query BtcPlusRewardByAddress($address: String!, $stageNo: Int!) { btcPlusRewardByAddress(address: $address, stageNo: $stageNo) { balance balanceUSD rewardPower currentTotalRewardPower estimatedReward __typename } } """ variables = {"stageNo": stage_no, "address": address} payload = { "operationName": "BtcPlusRewardByAddress", "query": query, "variables": variables, } response = requests.post(url, headers=headers, json=payload) response.raise_for_status() data = response.json() data["timestamp"] = get_current_time() # 添加当前时间 return data def fetch_btc_plus_allocations(): """从API获取BTC Plus分配数据""" url = "https://graphql.sft-api.com/graphql" headers = { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:141.0) Gecko/20100101 Firefox/141.0", "Accept": "*/*", "Accept-Language": "zh-CN,zh;q=0.8,en-US;q=0.5,en;q=0.3", "Accept-Encoding": "gzip, deflate, br, zstd", "Referer": "https://app.solv.finance/", "content-type": "application/json", "authorization": "dW5kZWZpbmVkfHx1bmRlZmluZWR8fHVuZGVmaW5lZHx8.undefined", "x-amz-user-agent": "aws-amplify/3.0.7", "Origin": "https://app.solv.finance", "DNT": "1", "Connection": "keep-alive", "Sec-Fetch-Dest": "empty", "Sec-Fetch-Mode": "cors", "Sec-Fetch-Site": "cross-site", "Sec-GPC": "1", "Priority": "u=4", "Pragma": "no-cache", "Cache-Control": "no-cache", "TE": "trailers", } query = """ query BtcPlusAllocations { btcPlusAllocations { tvl allocations { assetName percentage color __typename } __typename } } """ payload = { "operationName": "BtcPlusAllocations", "query": query, "variables": {}, } response = requests.post(url, headers=headers, json=payload) response.raise_for_status() data = response.json() data["timestamp"] = get_current_time() # 添加当前时间 return data def fetch_btc_plus_stats(stage_no=1): """从API获取BTC Plus统计数据""" url = "https://graphql.sft-api.com/graphql" headers = { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:141.0) Gecko/20100101 Firefox/141.0", "Accept": "*/*", "Accept-Language": "zh-CN,zh;q=0.8,en-US;q=0.5,en;q=0.3", "Accept-Encoding": "gzip, deflate, br, zstd", "Referer": "https://app.solv.finance/", "content-type": "application/json", "authorization": "dW5kZWZpbmVkfHx1bmRlZmluZWR8fHVuZGVmaW5lZHx8.undefined", "x-amz-user-agent": "aws-amplify/3.0.7", "Origin": "https://app.solv.finance", "DNT": "1", "Connection": "keep-alive", "Sec-Fetch-Dest": "empty", "Sec-Fetch-Mode": "cors", "Sec-Fetch-Site": "cross-site", "Sec-GPC": "1", "Priority": "u=4", } query = """ query BtcPlusStats($stageNo: Int) { btcPlusStats(stageNo: $stageNo) { baseApy rewardApy tvl cap lastUpdatedTime startDate endDate currentTotalRewardPower totalRewards __typename } } """ variables = {"stageNo": stage_no} payload = { "operationName": "BtcPlusStats", "query": query, "variables": variables, } response = requests.post(url, headers=headers, json=payload) response.raise_for_status() data = response.json() data["timestamp"] = get_current_time() # 添加当前时间 return data def fetch_phase2_points(address): """从API获取Phase2积分系统数据""" url = "https://graphql.sft-api.com/graphql" headers = { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:141.0) Gecko/20100101 Firefox/141.0", "Accept": "*/*", "Accept-Language": "zh-CN,zh;q=0.8,en-US;q=0.5,en;q=0.3", "Accept-Encoding": "gzip, deflate, br, zstd", "Referer": "https://app.solv.finance/", "content-type": "application/json", "authorization": "dW5kZWZpbmVkfHx1bmRlZmluZWR8fHVuZGVmaW5lZHx8.undefined", "x-amz-user-agent": "aws-amplify/3.0.7", "Origin": "https://app.solv.finance", "DNT": "1", "Connection": "keep-alive", "Sec-Fetch-Dest": "empty", "Sec-Fetch-Mode": "cors", "Sec-Fetch-Site": "cross-site", "Sec-GPC": "1", "Priority": "u=4", } query = """ query Phase2PointSysAccountInfo($address: String) { phase2PointSysAccountInfo(address: $address) { address isRegistered seedUserInviteCode inviteCode inviteCount totalPointsEarned availablePoints isPointsAccelerationActive todayHoldingTVL todayHoldingAccelerationRatio nextLevelHoldingTVL nextLevelHoldingAccelerationRatio activityCards { type accelerationRatio startTime endTime __typename } isHighestLevel __typename } } """ variables = {"address": address} payload = { "operationName": "Phase2PointSysAccountInfo", "query": query, "variables": variables, } response = requests.post(url, headers=headers, json=payload) response.raise_for_status() data = response.json() data["timestamp"] = get_current_time() return data def save_to_csv(data, filename="btc_plus_rewards.csv"): """将数据保存到CSV文件""" logger = logging.getLogger(__name__) timestamp = data.get("timestamp", get_current_time()) if "btcPlusRewardByAddress" in data.get("data", {}): reward_data = data["data"]["btcPlusRewardByAddress"] row = { "timestamp": timestamp, "address": data.get("variables", {}).get("address", "unknown"), "stageNo": data.get("variables", {}).get("stageNo", "unknown"), "balance": reward_data["balance"], "balanceUSD": reward_data["balanceUSD"], "rewardPower": reward_data["rewardPower"], "currentTotalRewardPower": reward_data["currentTotalRewardPower"], "estimatedReward": reward_data["estimatedReward"], } filename = "btc_plus_rewards.csv" elif "btcPlusAllocations" in data.get("data", {}): allocation_data = data["data"]["btcPlusAllocations"] row = { "timestamp": timestamp, "tvl": allocation_data["tvl"], "allocations": ";".join( f"{a['assetName']}:{a['percentage']}" for a in allocation_data["allocations"] ), } filename = "btc_plus_allocations.csv" elif "phase2PointSysAccountInfo" in data.get("data", {}): points_data = data["data"]["phase2PointSysAccountInfo"] row = { "timestamp": timestamp, "address": points_data["address"], "isRegistered": points_data["isRegistered"], "inviteCode": points_data["inviteCode"], "inviteCount": points_data["inviteCount"], "totalPointsEarned": points_data["totalPointsEarned"], "availablePoints": points_data["availablePoints"], "isPointsAccelerationActive": points_data["isPointsAccelerationActive"], "todayHoldingTVL": points_data["todayHoldingTVL"], "todayHoldingAccelerationRatio": points_data["todayHoldingAccelerationRatio"], "nextLevelHoldingTVL": points_data["nextLevelHoldingTVL"], "nextLevelHoldingAccelerationRatio": points_data["nextLevelHoldingAccelerationRatio"], "isHighestLevel": points_data["isHighestLevel"], } filename = "phase2_points.csv" else: stats_data = data["data"]["btcPlusStats"] row = { "timestamp": timestamp, "stageNo": data.get("variables", {}).get("stageNo", "unknown"), "baseApy": stats_data["baseApy"], "rewardApy": stats_data["rewardApy"], "tvl": stats_data["tvl"], "cap": stats_data["cap"], "lastUpdatedTime": timestamp_to_datetime(stats_data["lastUpdatedTime"]), "startDate": timestamp_to_datetime(stats_data["startDate"]), "endDate": timestamp_to_datetime(stats_data["endDate"]), "currentTotalRewardPower": stats_data["currentTotalRewardPower"], "totalRewards": stats_data["totalRewards"], } filename = "btc_plus_stats.csv" # 写入CSV文件 try: with open(filename, "a", newline="") as csvfile: writer = csv.DictWriter(csvfile, fieldnames=row.keys()) # 如果是新文件,写入表头 if csvfile.tell() == 0: writer.writeheader() writer.writerow(row) except Exception as e: logger.error("保存到CSV时出错: %s", e) # 初始化日志 def setup_logging(): """配置日志记录,只记录错误信息""" logging.basicConfig( level=logging.WARNING, # 只记录WARNING及以上级别 format="%(asctime)s - %(levelname)s - %(message)s", handlers=[ logging.FileHandler(LOG_FILE, encoding="utf-8"), ], ) return logging.getLogger(__name__) def git_commit_and_push(): """自动提交并推送数据更新到Git仓库""" logger = logging.getLogger(__name__) try: subprocess.run(["git", "add", "."], check=True) subprocess.run( ["git", "commit", "-m", f"Auto update at {datetime.now()}"], check=True, ) subprocess.run(["git", "push"], check=True) logger.info("Git提交成功") except subprocess.CalledProcessError as e: logger.error("Git操作失败: %s", e) def main_loop(): """主循环,定期获取数据并更新""" logger = setup_logging() logger.info("BTC Plus数据收集程序启动") last_git_time = 0 while True: try: # 获取并保存所有数据 address = "0x1bFD4C5bBD6d3eabDeec1Ece32Fb2dFC8f28C6b0" stage_no = 1 reward_data = fetch_btc_plus_reward(address, stage_no) reward_data["variables"] = {"address": address, "stageNo": stage_no} save_to_csv(reward_data) allocation_data = fetch_btc_plus_allocations() save_to_csv(allocation_data) stats_data = fetch_btc_plus_stats(stage_no) stats_data["variables"] = {"stageNo": stage_no} save_to_csv(stats_data) points_data = fetch_phase2_points(address) save_to_csv(points_data) # 定时Git提交 current_time = time.time() if current_time - last_git_time >= GIT_INTERVAL: git_commit_and_push() last_git_time = current_time time.sleep(FETCH_INTERVAL) except Exception as e: logger.error("发生错误: %s,10秒后重试...", e) time.sleep(10) if __name__ == "__main__": main_loop()