"""DEXRP价格数据收集与分析模块,自动获取交易数据并生成每日统计""" import json import time import subprocess from datetime import datetime, timezone from collections import defaultdict import requests # 配置常量 API_URL = "https://backend.dexrp.io/vending/last" # DEXRP API地址 JSON_FILE = "price.json" # 价格数据存储文件 HASH_FILE = "processed.json" # 已处理交易哈希记录文件 # 全局变量 SEEN_TXHASHES = set() # 内存中的已处理交易哈希集合 GIT_INTERVAL = 300 # Git提交间隔(秒) FETCH_INTERVAL = 10 # API轮询间隔(秒) def fetch_latest_transactions(): """从API获取最新交易数据并过滤已处理记录""" try: response = requests.get(API_URL, timeout=10) response.raise_for_status() transactions = response.json() # transactions = json.loads( # open("D:/DOWNLOADS/last.json", "r", encoding="utf-8").read() # ) return [tx for tx in transactions if tx["transactionHash"] not in SEEN_TXHASHES] except requests.RequestException as e: print(f"API请求失败: {e}") return None def calculate_daily_stats(transactions): """计算每日统计数据(开盘价、收盘价、最高价、最低价、交易量)""" daily_data = defaultdict( lambda: { "open": None, "close": None, "high": float("-inf"), "low": float("inf"), "tokensSold": 0, } ) for tx in transactions: date = datetime.fromtimestamp(tx["blockTimestamp"], timezone.utc).strftime("%Y-%m-%d") price = tx.get("price", 0) tokens_sold = tx.get("tokensSold", 0) if daily_data[date]["open"] is None: daily_data[date]["open"] = price daily_data[date]["close"] = price daily_data[date]["high"] = max(daily_data[date]["high"], price) daily_data[date]["low"] = min(daily_data[date]["low"], price) daily_data[date]["tokensSold"] += tokens_sold return [ { "date": date, "open": data["open"], "close": data["close"], "high": data["high"], "low": data["low"], "tokensSold": data["tokensSold"], } for date, data in daily_data.items() ] def update_json_file(new_data): """Update JSON file with new transaction data.""" try: print(f"开始更新JSON文件,收到{len(new_data)}条新交易数据") # 读取现有数据,如果文件不存在则初始化为空数组 try: with open(JSON_FILE, "r", encoding="utf-8") as file: existing_data = json.load(file) except FileNotFoundError: existing_data = [] # 确保existing_data是列表 if not isinstance(existing_data, list): existing_data = [] # 计算每日统计数据 processed_data = calculate_daily_stats(new_data) # 合并数据而不是追加 existing_dates = {item["date"]: item for item in existing_data} for new_item in processed_data: date = new_item["date"] if date in existing_dates: # 合并相同日期的记录 existing_item = existing_dates[date] existing_item["close"] = new_item["close"] existing_item["high"] = max(existing_item["high"], new_item["high"]) existing_item["low"] = min(existing_item["low"], new_item["low"]) existing_item["tokensSold"] += new_item["tokensSold"] else: existing_data.append(new_item) existing_dates[date] = new_item # 写回文件 with open(JSON_FILE, "w", encoding="utf-8") as file: json.dump(existing_data, file, indent=4, ensure_ascii=False) print( f"成功更新{JSON_FILE},合并{len(processed_data)}条记录,总计{len(existing_data)}条记录" ) except (FileNotFoundError, json.JSONDecodeError, IOError) as e: print(f"更新JSON文件时发生错误: {e},跳过本次更新") def git_commit_and_push(): """自动提交并推送数据更新到Git仓库""" try: subprocess.run(["git", "add", "."], check=True) subprocess.run( [ "git", "commit", "--no-verify", "--no-gpg-sign", "-m", f"Auto update at {datetime.now()}", ], check=True, ) subprocess.run(["git", "push"], check=True) print("Git提交成功") except subprocess.CalledProcessError as e: print(f"Git操作失败: {e}") def load_processed_hashes(): """Load processed transaction hashes from file.""" try: with open(HASH_FILE, "r", encoding="utf-8") as file: return set(json.load(file)) except (FileNotFoundError, json.JSONDecodeError): return set() def save_processed_hashes(): """Save processed transaction hashes to file.""" with open(HASH_FILE, "w", encoding="utf-8") as file: json.dump(list(SEEN_TXHASHES), file) def main(): """主循环,定期获取数据并更新""" global SEEN_TXHASHES SEEN_TXHASHES = load_processed_hashes() last_git_time = 0 while True: try: current_time = time.time() print(f"{datetime.now()}: 获取最新交易数据...") if new_transactions := fetch_latest_transactions(): print(f"获取到{len(new_transactions)}条新交易") SEEN_TXHASHES.update(tx["transactionHash"] for tx in new_transactions) save_processed_hashes() update_json_file(new_transactions) 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: print(f"错误: {e},继续运行...") continue if __name__ == "__main__": main()