[Python] How can I perform bulk insertion combined with `ON CONFLICT DO NOTHING`?

I have a substantial collection of financial market data in multiple CSV files that I need to store in a database for future use in my software.

I am employing Python with SQLAlchemy and SQLite.

Here's what my data looks like, along with my database models:

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<id={self.id!r}, date={self.date!r}>"


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<id={self.id!r}, time={self.time!r}>"


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<id={self.id!r}, name={self.name!r}, markets={self.markets!r}>"

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<id={self.id!r}, name={self.name!r}, region={self.region!r}, assets={self.assets!r}>"


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<id={self.id!r}, symbol={self.symbol!r}, name={self.name!r}, market={self.market!r}>"


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<id={self.id!r}, date={self.date!r}, time={self.time!r}, asset={self.asset!r}, opening={self.opening!r}, high={self.high!r}, low={self.low!r}, closing={self.closing!r}, volume={self.volume!r}>"



I utilized the following code for the initial dates insertion.

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()


However, this dataset is regularly updated, and I need to insert new data into the database while ensuring data integrity, which currently is too slow.

This is what I tried to do for inserting new data into the database.

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()


How can I perform bulk insertion combined with ON CONFLICT DO NOTHING? I think this is the most efficient method I can use.

Continue reading...