如何将Pandas DataFrame更新到PostgreSQL多表的现有行（按列匹配）

背景 #

数据库存在三个独立PostgreSQL表：AUDCAD、AUDJPY、AUDUSD，需将MACD相关NumPy数组数据更新到各表原有行的macd_480、macd_840、macd_1518、macd_2400列，而非新增行。原INSERT逻辑会生成空值行，UPDATE执行报错，以下提供psycopg2和SQLAlchemy两种解决方案。

数据示例 #

更新前表数据 #

row_no    O   H   L   C  macd_480  macd_840  macd_1518  macd_2400
0      1    10  20  30  40     None      None        None      None
1      2    10  20  30  40     None      None        None      None
2      3    10  20  30  40     None      None        None      None
3      4    10  20  30  40     None      None        None      None

INSERT后（不符合预期） #

row_no     O     H     L     C  macd_480  macd_840  macd_1518  macd_2400
0       1     10    20    30    40     None      None       None       None
1       2     10    20    30    40     None      None       None       None
2       3     10    20    30    40     None      None       None       None
3       4     10    20    30    40     None      None       None       None
4       5   None  None  None  None       50        60         70         80
5       6   None  None  None  None       50        60         70         80
6       7   None  None  None  None       50        60         70         80
7       8   None  None  None  None       50        60         70         80

预期结果 #

row_no   O   H   L   C  macd_480  macd_840  macd_1518  macd_2400
0       1   10  20  30  40       50        60         70         80
1       2   10  20  30  40       50        60         70         80
2       3   10  20  30  40       50        60         70         80
3       4   10  20  30  40       50        60         70         80

方案一：psycopg2批量更新 #

核心是利用PostgreSQL的UPDATE...FROM语法，将MACD数据与表中row_no关联，批量匹配更新，避免单条更新效率低下。

import pandas as pd
import numpy as np
import psycopg2
from psycopg2 import extras

# 数据库连接信息，替换为你的实际配置
DB_CONFIG = {
    "dbname": "your_db",
    "user": "your_user",
    "password": "your_pwd",
    "host": "your_host"
}

SYMBOLS = ['AUDCAD', 'AUDJPY', 'AUDUSD']

# 生成带row_no的数据集（row_no与表中现有行一一对应）
def prepare_update_data(macd_480, macd_840, macd_1518, macd_2400):
    macd_df = pd.DataFrame(
        np.column_stack((macd_480, macd_840, macd_1518, macd_2400)),
        columns=['macd_480', 'macd_840', 'macd_1518', 'macd_2400']
    ).drop(index=0)
    # 匹配表中row_no（drop(index=0)后索引为1-4，对应表中row_no 1-4）
    macd_df['row_no'] = macd_df.index
    return macd_df[['row_no', 'macd_480', 'macd_840', 'macd_1518', 'macd_2400']].values.tolist()

# 批量更新SQL模板
UPDATE_QUERY = """
UPDATE {table}
SET macd_480 = data.macd_480,
    macd_840 = data.macd_840,
    macd_1518 = data.macd_1518,
    macd_2400 = data.macd_2400
FROM (VALUES %s) AS data(row_no, macd_480, macd_840, macd_1518, macd_2400)
WHERE {table}.row_no = data.row_no;
"""

# 执行更新
conn = psycopg2.connect(**DB_CONFIG)
cursor = conn.cursor()

for symbol in SYMBOLS:
    # 替换为当前symbol对应的MACD数组
    update_data = prepare_update_data(macd_480, macd_840, macd_1518, macd_2400)
    query = UPDATE_QUERY.format(table=symbol)
    extras.execute_values(
        cursor,
        query,
        update_data,
        template="(%s, %s, %s, %s, %s)",
        page_size=1000
    )

conn.commit()
cursor.close()
conn.close()

方案二：SQLAlchemy批量更新 #

利用SQLAlchemy的bulk_update_mappings方法，结合Pandas处理数据，代码更简洁，无需手写复杂SQL。

from sqlalchemy import create_engine, Table, MetaData
import pandas as pd
import numpy as np

# 数据库连接字符串，替换为你的实际配置
DB_URL = "postgresql://your_user:your_pwd@your_host:5432/your_db"
engine = create_engine(DB_URL)
metadata = MetaData()

SYMBOLS = ['AUDCAD', 'AUDJPY', 'AUDUSD']

# 生成用于批量更新的字典列表
def prepare_update_mappings(macd_480, macd_840, macd_1518, macd_2400):
    macd_df = pd.DataFrame(
        np.column_stack((macd_480, macd_840, macd_1518, macd_2400)),
        columns=['macd_480', 'macd_840', 'macd_1518', 'macd_2400']
    ).drop(index=0)
    macd_df['row_no'] = macd_df.index
    return macd_df.to_dict('records')

# 执行批量更新
with engine.begin() as conn:
    for symbol in SYMBOLS:
        # 加载表结构
        table = Table(symbol, metadata, autoload_with=engine)
        # 替换为当前symbol对应的MACD数组
        update_mappings = prepare_update_mappings(macd_480, macd_840, macd_1518, macd_2400)
        # 批量更新
        conn.bulk_update_mappings(table, update_mappings)

内容的提问来源于stack exchange，提问作者rn kim