118 lines
4.3 KiB
Python
118 lines
4.3 KiB
Python
|
import requests
|
|||
|
|
import pymysql
|
|||
|
|
from datetime import datetime
|
|||
|
|
import time
|
|||
|
|
|
|||
|
|
BEA_USER_ID = "146B5757-D9E3-442C-B6AC-ADE9E6B71114"
|
|||
|
|
YEARS = ["2023","2024","2025"] # 第一次运行抓全部年份
|
|||
|
|
SLEEP_SECONDS = 21600 # 6小时
|
|||
|
|
|
|||
|
|
def get_bea_data(year):
|
|||
|
|
"""抓取指定年份的季度数据"""
|
|||
|
|
url = (
|
|||
|
|
f'https://apps.bea.gov/api/data?UserID={BEA_USER_ID}'
|
|||
|
|
f'&method=GetData&datasetname=NIPA&TableName=T10105&Frequency=Q'
|
|||
|
|
f'&Year={year}&ResultFormat=JSON'
|
|||
|
|
)
|
|||
|
|
response = requests.get(url)
|
|||
|
|
return response.json()['BEAAPI']['Results']['Data']
|
|||
|
|
|
|||
|
|
def update_database(cursor, data):
|
|||
|
|
"""整理并插入缺失季度数据"""
|
|||
|
|
# 查询数据库已存在的季度
|
|||
|
|
cursor.execute("SELECT times FROM PCE")
|
|||
|
|
existing_pce = {row[0] for row in cursor.fetchall()}
|
|||
|
|
cursor.execute("SELECT times FROM GPDI")
|
|||
|
|
existing_gpdi = {row[0] for row in cursor.fetchall()}
|
|||
|
|
cursor.execute("SELECT times FROM NETEXP")
|
|||
|
|
existing_netexp = {row[0] for row in cursor.fetchall()}
|
|||
|
|
|
|||
|
|
# 整理数据
|
|||
|
|
pce_data, gpdi_data, netexp_data = {}, {}, {}
|
|||
|
|
for entry in data:
|
|||
|
|
t = entry["TimePeriod"]
|
|||
|
|
desc = entry["LineDescription"]
|
|||
|
|
val = entry["DataValue"]
|
|||
|
|
|
|||
|
|
if desc == "Personal consumption expenditures":
|
|||
|
|
pce_data.setdefault(t, {})["PCE"] = val
|
|||
|
|
elif desc == "Durable goods":
|
|||
|
|
pce_data.setdefault(t, {})["PCEDG"] = val
|
|||
|
|
elif desc == "Nondurable goods":
|
|||
|
|
pce_data.setdefault(t, {})["PCEND"] = val
|
|||
|
|
elif desc == "Services" and entry["LineNumber"] == '6':
|
|||
|
|
pce_data.setdefault(t, {})["PCES"] = val
|
|||
|
|
|
|||
|
|
elif desc == "Gross private domestic investment":
|
|||
|
|
gpdi_data.setdefault(t, {})["GPDI"] = val
|
|||
|
|
elif desc == "Fixed investment":
|
|||
|
|
gpdi_data.setdefault(t, {})["FPI"] = val
|
|||
|
|
elif desc == "Change in private inventories":
|
|||
|
|
gpdi_data.setdefault(t, {})["CBI"] = val
|
|||
|
|
|
|||
|
|
elif desc == "Net exports of goods and services":
|
|||
|
|
netexp_data.setdefault(t, {})["NETEXP"] = val
|
|||
|
|
elif desc == "Imports":
|
|||
|
|
netexp_data.setdefault(t, {})["IMPGS"] = val
|
|||
|
|
elif desc == "Exports":
|
|||
|
|
netexp_data.setdefault(t, {})["EXPGS"] = val
|
|||
|
|
|
|||
|
|
# 插入数据库缺失数据
|
|||
|
|
for t, vals in pce_data.items():
|
|||
|
|
if t not in existing_pce:
|
|||
|
|
cursor.execute(
|
|||
|
|
"INSERT INTO PCE (times, PCE, PCEDG, PCEND, PCES) VALUES (%s,%s,%s,%s,%s)",
|
|||
|
|
(t, vals.get("PCE"), vals.get("PCEDG"), vals.get("PCEND"), vals.get("PCES"))
|
|||
|
|
)
|
|||
|
|
for t, vals in gpdi_data.items():
|
|||
|
|
if t not in existing_gpdi:
|
|||
|
|
cursor.execute(
|
|||
|
|
"INSERT INTO GPDI (times, GPDI, FPI, CBI) VALUES (%s,%s,%s,%s)",
|
|||
|
|
(t, vals.get("GPDI"), vals.get("FPI"), vals.get("CBI"))
|
|||
|
|
)
|
|||
|
|
for t, vals in netexp_data.items():
|
|||
|
|
if t not in existing_netexp:
|
|||
|
|
cursor.execute(
|
|||
|
|
"INSERT INTO NETEXP (times, NETEXP, IMPGS, EXPGS) VALUES (%s,%s,%s,%s)",
|
|||
|
|
(t, vals.get("NETEXP"), vals.get("IMPGS"), vals.get("EXPGS"))
|
|||
|
|
)
|
|||
|
|
|
|||
|
|
def run_job(first_run=False):
|
|||
|
|
"""运行一次抓取和更新"""
|
|||
|
|
print(f"[{datetime.now()}] 开始抓取 BEA 数据并更新数据库...")
|
|||
|
|
try:
|
|||
|
|
db = pymysql.connect(
|
|||
|
|
host="127.0.0.1",
|
|||
|
|
user="root",
|
|||
|
|
password="2GS@bPYcgiMyL14A",
|
|||
|
|
database="Macroeconomics",
|
|||
|
|
port=4423
|
|||
|
|
)
|
|||
|
|
cursor = db.cursor()
|
|||
|
|
|
|||
|
|
years_to_fetch = YEARS if first_run else [YEARS[-1]] # 第一次抓全部年份，否则只抓最新年份
|
|||
|
|
for year in years_to_fetch:
|
|||
|
|
data = get_bea_data(year)
|
|||
|
|
update_database(cursor, data)
|
|||
|
|
db.commit()
|
|||
|
|
print(f"[{datetime.now()}] {year} 数据更新完成")
|
|||
|
|
|
|||
|
|
except pymysql.MySQLError as e:
|
|||
|
|
print(f"[{datetime.now()}] 数据库错误: {e}")
|
|||
|
|
except Exception as e:
|
|||
|
|
print(f"[{datetime.now()}] 其他错误: {e}")
|
|||
|
|
finally:
|
|||
|
|
if 'cursor' in locals():
|
|||
|
|
cursor.close()
|
|||
|
|
if 'db' in locals():
|
|||
|
|
db.close()
|
|||
|
|
print(f"[{datetime.now()}] 本次任务完成。\n")
|
|||
|
|
|
|||
|
|
if __name__ == "__main__":
|
|||
|
|
first_run = True
|
|||
|
|
while True:
|
|||
|
|
run_job(first_run)
|
|||
|
|
first_run = False # 之后循环只抓最新季度
|
|||
|
|
print(f"[{datetime.now()}] 休眠 {SLEEP_SECONDS} 秒（6小时）...\n")
|
|||
|
|
time.sleep(SLEEP_SECONDS)
|