98 lines
2.9 KiB
Python
98 lines
2.9 KiB
Python
|
import requests
|
|||
|
|
import pymysql
|
|||
|
|
from datetime import datetime
|
|||
|
|
from w3lib.html import remove_tags
|
|||
|
|
import pandas as pd
|
|||
|
|
import time
|
|||
|
|
|
|||
|
|
def parse_treasury_data(data):
|
|||
|
|
# 找到列头位置
|
|||
|
|
header_index = data.index("Country")
|
|||
|
|
columns = data[header_index:header_index+14] # Country + 13个月
|
|||
|
|
rows = data[header_index+14:]
|
|||
|
|
|
|||
|
|
result = []
|
|||
|
|
i = 0
|
|||
|
|
while i < len(rows):
|
|||
|
|
# 拼接国家名
|
|||
|
|
country_parts = []
|
|||
|
|
while i < len(rows) and not rows[i].replace('.', '', 1).isdigit():
|
|||
|
|
country_parts.append(rows[i])
|
|||
|
|
i += 1
|
|||
|
|
country = " ".join(country_parts).replace(",", "")
|
|||
|
|
|
|||
|
|
# 取13个数值
|
|||
|
|
values = rows[i:i+13]
|
|||
|
|
i += 13
|
|||
|
|
|
|||
|
|
if len(values) == 13:
|
|||
|
|
result.append([country] + values)
|
|||
|
|
|
|||
|
|
# 转成 DataFrame
|
|||
|
|
df = pd.DataFrame(result, columns=columns)
|
|||
|
|
|
|||
|
|
# =================== 名称清洗 ===================
|
|||
|
|
rename_map = {
|
|||
|
|
"Of Which: Foreign Official": "Foreign Official",
|
|||
|
|
"Of Which: Foreign Official Treasury Bills": "Treasury Bills",
|
|||
|
|
"Of Which: Foreign Official T-Bonds & Notes": "T-Bonds & Notes"
|
|||
|
|
}
|
|||
|
|
df["Country"] = df["Country"].replace(rename_map)
|
|||
|
|
|
|||
|
|
return df
|
|||
|
|
|
|||
|
|
|
|||
|
|
def run_job():
|
|||
|
|
print("=== 开始爬取并更新数据库 ===")
|
|||
|
|
|
|||
|
|
# =================== 爬取网页 =====================
|
|||
|
|
page = requests.get("https://ticdata.treasury.gov/resource-center/data-chart-center/tic/Documents/slt_table5.html")
|
|||
|
|
page = remove_tags(str(page.text))
|
|||
|
|
page = page.split()
|
|||
|
|
|
|||
|
|
df = parse_treasury_data(page)
|
|||
|
|
|
|||
|
|
# =================== 连接数据库 =====================
|
|||
|
|
db = pymysql.connect(
|
|||
|
|
host="127.0.0.1",
|
|||
|
|
user="root",
|
|||
|
|
password="2GS@bPYcgiMyL14A",
|
|||
|
|
database="Macroeconomics",
|
|||
|
|
port=4423
|
|||
|
|
)
|
|||
|
|
cursor = db.cursor()
|
|||
|
|
|
|||
|
|
# 查询数据库中最新日期
|
|||
|
|
cursor.execute("SELECT date FROM FBI ORDER BY date DESC LIMIT 1")
|
|||
|
|
result = cursor.fetchone()
|
|||
|
|
latest_date_in_db = result[0] if result else None # datetime 类型或 None
|
|||
|
|
|
|||
|
|
# =================== 补齐逻辑 =====================
|
|||
|
|
for col in df.columns[1:]: # 遍历所有月份列
|
|||
|
|
col_date = datetime.strptime(col, "%Y-%m")
|
|||
|
|
|
|||
|
|
# 如果数据库已有该日期，跳过
|
|||
|
|
if latest_date_in_db and col_date <= latest_date_in_db:
|
|||
|
|
continue
|
|||
|
|
|
|||
|
|
print(f"正在插入 {col} 的数据...")
|
|||
|
|
insert_sql = "INSERT INTO FBI (date, name, value) VALUES (%s, %s, %s)"
|
|||
|
|
for _, row in df.iterrows():
|
|||
|
|
country = row["Country"]
|
|||
|
|
value = row[col]
|
|||
|
|
cursor.execute(insert_sql, (col_date.strftime("%Y-%m-01"), country, value))
|
|||
|
|
|
|||
|
|
db.commit()
|
|||
|
|
print(f"{col} 插入完成")
|
|||
|
|
|
|||
|
|
cursor.close()
|
|||
|
|
db.close()
|
|||
|
|
print("=== 本次任务完成 ===\n")
|
|||
|
|
|
|||
|
|
|
|||
|
|
# =================== 循环执行 =====================
|
|||
|
|
if __name__ == "__main__":
|
|||
|
|
while True:
|
|||
|
|
run_job()
|
|||
|
|
print("休眠 21600 秒（6 小时）...\n")
|
|||
|
|
time.sleep(21600) # 6小时
|