用 Python 中的每日股票价格更新 PostgreSQL 数据库

Question

因此，我在 QuantState 找到了一个很棒的脚本，它对设置我自己的证券数据库和加载历史定价信息进行了很好的演练。但是，我并不想修改脚本，以便我可以每天运行它并添加最新的股票报价。

我将初始数据加载调整为仅下载 1 周的历史数据，但我在编写 SQL 语句以查看该行是否在添加之前已经存在时遇到了问题。谁能帮我解决这个问题。这是我目前所拥有的：

def insert_daily_data_into_db(data_vendor_id, symbol_id, daily_data):
  """Takes a list of tuples of daily data and adds it to the
   database. Appends the vendor ID and symbol ID to the data.

  daily_data: List of tuples of the OHLC data (with 
  adj_close and volume)"""

  # Create the time now
  now = datetime.datetime.utcnow()

  # Amend the data to include the vendor ID and symbol ID
  daily_data = [(data_vendor_id, symbol_id, d[0], now, now,
    d[1], d[2], d[3], d[4], d[5], d[6]) for d in daily_data]

  # Create the insert strings
  column_str = """data_vendor_id, symbol_id, price_date, created_date, 
          last_updated_date, open_price, high_price, low_price, 
          close_price, volume, adj_close_price"""
  insert_str = ("%s, " * 11)[:-2]
  final_str = "INSERT INTO daily_price (%s) VALUES (%s) WHERE NOT EXISTS (SELECT 1 FROM daily_price WHERE symbol_id = symbol_id AND price_date = insert_str[2])" % (column_str, insert_str)

  # Using the postgre connection, carry out an INSERT INTO for every symbol
  with con: 
    cur = con.cursor()
    cur.executemany(final_str, daily_data)

Answer 1

关于上面代码的一些注释：

在纯SQL 中遵循now() 通常比在Python 中尝试更容易。它避免了许多潜在的时区、库差异等问题。

如果你构造一个列的list，你可以根据它的大小动态生成一个%s的字符串，并且不需要将长度硬编码成一个重复的字符串然后切片。

由于insert_daily_data_into_db 似乎是要在每个股票的循环内调用，我不相信你想在这里使用executemany，这需要一个列表 em> 的 元组 并且在语义上非常不同。

您在子 select 中将 symbol_id 与自身进行比较，而不是特定值（这意味着它始终为真）。

为防止可能的SQL 注入，您应该始终在WHERE 子句中插入值，包括子selects。

注意：我假设您正在使用 psycopg2 访问 Postgres，并且该表的主键是(symbol_id, price_date) 的元组。如果没有，下面的代码至少需要稍微调整一下。

考虑到这些点，尝试这样的事情（未经测试，因为我没有你的数据、数据库等，但它在语法上是有效的 Python）：

def insert_daily_data_into_db(data_vendor_id, symbol_id, daily_data):
    """Takes a list of tuples of daily data and adds it to the
    database. Appends the vendor ID and symbol ID to the data.

    daily_data: List of tuples of the OHLC data (with
    adj_close and volume)"""


    column_list = ["data_vendor_id", "symbol_id", "price_date", "created_date",
                    "last_updated_date", "open_price", "high_price", "low_price",
                    "close_price", "volume", "adj_close_price"]

    insert_list = ['%s'] * len(column_str)

    values_tuple = (data_vendor_id, symbol_id, daily_data[0], 'now()', 'now()', daily_data[1],
                    daily_data[2], daily_data[3], daily_data[4], daily_data[5], daily_data[6])

    final_str = """INSERT INTO daily_price ({0})
                     VALUES ({1})
                     WHERE NOT EXISTS (SELECT 1
                                       FROM daily_price
                                       WHERE symbol_id = %s
                                         AND price_date = %s)""".format(', '.join(column_list), ', '.join(insert_list))

    # Using the postgre connection, carry out an INSERT INTO for every symbol
    with con:
        cur = con.cursor()
        cur.execute(final_str, values_tuple, values_tuple[1], values_tuple[2])