Как я могу выполнить массовую вставку в сочетании с «ON CONFLICT DO NOTHING»? ⇐ Python

[Anonymous]

У меня есть значительная коллекция данных финансового рынка в нескольких файлах CSV, которые мне нужно сохранить в базе данных для будущего использования в моем программном обеспечении.
Я использую Python с SQLAlchemy и SQLite .
Вот как выглядят мои данные вместе с моделями моей базы данных:

Код: Выделить всё

            Open    High    Low     Close
Timestamp
31798800    0.53690 0.53690 0.53690 0.53690
31802400    0.53690 0.53690 0.53690 0.53690
31885200    0.53660 0.53660 0.53660 0.53660
31888800    0.53660 0.53660 0.53660 0.53660
31971600    0.53650 0.53650 0.53650 0.53650
... ... ... ...  ...
1731113460  1.07177 1.07185 1.07174 1.07182
1731113520  1.07182 1.07182 1.07177 1.07178
1731113580  1.07179 1.07187 1.07178 1.07186
1731113640  1.07186 1.07186 1.07178 1.07180
1731113700  1.07181 1.07181 1.07181 1.07181

Код: Выделить всё

import datetime
from typing import List
from sqlalchemy import ForeignKey
from sqlalchemy import String, Integer, Float, Date, Time
from sqlalchemy.orm import DeclarativeBase
from sqlalchemy.orm import Mapped
from sqlalchemy.orm import mapped_column
from sqlalchemy.orm import relationship

class Base(DeclarativeBase):
pass

class Date(Base):
__tablename__ = "dates"
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True, unique=True)
date: Mapped[datetime.date] = mapped_column(Date, nullable=False, unique=True)

def __repr__(self) -> str:
return f"Date"

class Time(Base):
__tablename__ = "times"
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True, unique=True)
time: Mapped[datetime.time] = mapped_column(Time, nullable=False, unique=True)

def __repr__(self) -> str:
return f"Time"

class Region(Base):
__tablename__ = "regions"
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True, unique=True)
name: Mapped[str] = mapped_column(String(30), nullable=False, unique=True)
markets: Mapped[List["Market"]] = relationship()

def __repr__(self) -> str:
return f"Region"

class Market(Base):
__tablename__ = "markets"
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True, unique=True)
name: Mapped[str] = mapped_column(String(20), nullable=False)
region: Mapped[int] = mapped_column(ForeignKey("regions.id"))
assets: Mapped[List["Asset"]] = relationship()

def __repr__(self) -> str:
return f"Market"

class Asset(Base):
__tablename__ = "assets"
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True, unique=True)
symbol: Mapped[str] = mapped_column(String(10), nullable=False, unique=True)
name: Mapped[str] = mapped_column(String(30), nullable=True)
market: Mapped[int] = mapped_column(ForeignKey("markets.id"))

def __repr__(self) -> str:
return f"Asset"

class MarketData(Base):
__tablename__ = "market_data"
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True, unique=True)
date: Mapped[int] = mapped_column(ForeignKey("dates.id"))
time: Mapped[int] = mapped_column(ForeignKey("times.id"))
asset: Mapped[int] = mapped_column(ForeignKey("assets.id"))
opening: Mapped[float] = mapped_column(Float, nullable=False)
high: Mapped[float] = mapped_column(Float, nullable=False)
low: Mapped[float] = mapped_column(Float, nullable=False)
closing: Mapped[float] = mapped_column(Float, nullable=False)
volume: Mapped[float] = mapped_column(Float, nullable=True)

def __repr__(self) ->  str:
return f"Market Value"

Для начальной вставки дат я использовал следующий код.

Код: Выделить всё

symbol = "EURUSD"
filenames = get_symbol_data_filenames(symbol)
total_df = None
for filename in filenames:
df = read_csv(filename)
df = df.set_index(["Timestamp"])
if total_df is None:
total_df = df
else:
total_df = concat([df, total_df])
total_df = total_df.sort_index()
total_df = total_df[~total_df.index.duplicated(keep="last")]
database_engine = create_engine("sqlite:///data/db.sqlite3")
Base.metadata.create_all(database_engine)
with Session(database_engine) as session:
global_region = session.query(Region).filter_by(name="Global").first()
if global_region is None:
global_region = Region(name="Global")
session.add(global_region)
session.commit()
forex = session.query(Market).filter_by(name="Forex", region=global_region.id).first()
if forex is None:
forex = Market(name="Forex", region=global_region.id)
session.add(forex)
session.commit()
eurusd = session.query(Asset).filter_by(symbol="EURUSD", name="Euro/US-Dollar", market=forex.id).first()
if eurusd is None:
eurusd = Asset(symbol="EURUSD", name="Euro/US-Dollar", market=forex.id)
session.add(eurusd)
session.commit()
dates = [date.fromtimestamp(ts) for ts in dft.index]
seen = set()
dates = [x for x in dates if x not in seen and not seen.add(x)]
session.bulk_insert_mappings(Date, [{"date": date_} for date_ in dates])
session.commit()

Однако этот набор данных регулярно обновляется, и мне нужно вставлять новые данные в базу данных, обеспечивая при этом целостность данных, что в настоящее время происходит слишком медленно.
Это то, что я пытался сделать для вставки новых данных в базу данных.

Код: Выделить всё

with Session(database_engine) as session:
dates = session.query(Date).all()
if dates is None:
new_dates = [date.fromtimestamp(ts) for ts in total_df.index]
seen = set()
new_dates = [x for x in new_dates if x not in seen and not seen.add(x)]
session.bulk_insert_mappings(Date, [{"date": date_} for date_ in new_dates])
session.commit()
else:
dates = [x.date for x in dates]
new_dates = [date.fromtimestamp(ts) for ts in total_df.index]
seen = set()
new_dates = [x for x in new_dates if x not in seen and x not in dates and not seen.add(x)]
session.bulk_insert_mappings(Date, [{"date": date_} for date_ in new_dates])
session.commit()

Как выполнить массовую вставку в сочетании с ON CONFLICT DO NOTHING? Я думаю, что это самый эффективный метод, который я могу использовать.

Подробнее здесь: https://stackoverflow.com/questions/792 ... do-nothing