当前位置：首页 > news >正文

python炒股

news 2025/9/7 6:35:13

python 获取股票历史数据存入mysql

# -*- coding: gbk -*-

import akshare as ak
import pymysql
from datetime import datetime, timedelta

def get_previous_trading_day():
today = datetime.now()
# 简单实现获取上一个交易日（实际应排除周末和节假日）
return (today - timedelta(days=1)).strftime("%Y%m%d")

def store_stock_data(symbol, months=6):
try:
# 创建数据库连接
mydb = pymysql.connect(
host="mysql.sqlpub.com",
user="cybroot",
password="uf3HbUuawr17GtmF",
database="cyb_database",
port=3306,
charset='utf8mb4',
cursorclass=pymysql.cursors.DictCursor
)
mycursor = mydb.cursor()
# 准备表结构
create_table_sql = f"""
CREATE TABLE IF NOT EXISTS `trading_daily_data` (
`date` DATE NOT NULL,
`open` DECIMAL(10,2),
`close` DECIMAL(10,2),
`high` DECIMAL(10,2),
`low` DECIMAL(10,2),
`volume` BIGINT,
`amount` DECIMAL(15,2),
`amplitude` DECIMAL(5,2),
`change_percent` DECIMAL(5,2),
`change_amount` DECIMAL(10,2),
`turnover_rate` DECIMAL(5,2),
PRIMARY KEY (`date`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
"""
#mycursor.execute(create_table_sql)

# 计算半年前的日期
end_date = get_previous_trading_day()
start_date = (datetime.now() - timedelta(days=30*months)).strftime("%Y%m%d")

current_date = start_date
batch_size = 5 # 每次处理5条数据

while current_date < end_date:
# 获取5条数据
df = ak.stock_zh_a_hist(
symbol=symbol,
period="daily",
start_date=current_date,
end_date=end_date
).head(batch_size)

if df.empty:
break

# 存储数据到MySQL
for _, row in df.iterrows():
sql = f"""
INSERT INTO trading_daily_data
(date, open, close, high, low, volume, amount, amplitude, change_percent, change_amount, turnover_rate, code)
VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
"""
val = (
row['日期'], row['开盘'], row['收盘'],
row['最高'], row['最低'], row['成交量'],
row['成交额'], row['振幅'], row['涨跌幅'], row['涨跌额'], row['换手率'], row['股票代码']
)
mycursor.execute(sql, val)

mydb.commit()
print(f"已存储{len(df)}条数据，日期范围: {df['日期'].min()} 至 {df['日期'].max()}")

# 修正时间处理部分
if not df.empty:
last_date_str = df['日期'].max().strftime("%Y%m%d")
last_date = datetime.strptime(last_date_str, "%Y%m%d")
current_date = (last_date + timedelta(days=1)).strftime("%Y%m%d")

except Exception as e:
print(f"发生错误: {str(e)}")
finally:
if 'mydb' in locals() and mydb:
mycursor.close()
mydb.close()

if __name__ == "__main__":
store_stock_data(symbol="600562", months=6)
------------------------------------------------------------------------------------------------------------