Skip to content

Database API Reference

SQLite

financelib.database.sqlite.SQLite(db_name: str)

SQLite database wrapper for FinanceLib data storage.

Creates a date-stamped database file and provides methods for inserting and querying news articles and stock data.

Parameters:

Name Type Description Default
db_name str

Logical name for the database (used in filename).

 required

Attributes:

Name Type Description
filename

Full database filename with date stamp.
db_name

Logical database name.
conn

SQLite connection.
cursor

SQLite cursor.
Example

db = SQLite("market_data") db.insert_article(news_model) db.close()

Source code in financelib/database/sqlite.py 
45
46
47
48
49
50
51
52
def __init__(self, db_name: str) -> None:
    self.filename = f"{LIBRARY_NAME}_{db_name}_{today_str_underline()}.db"
    self.db_name = db_name
    self.conn = sqlite3.connect(self.filename, check_same_thread=False)
    self.cursor = self.conn.cursor()
    self._in_transaction = False
    self._initialize_tables()
    logger.info(f"Database initialized: {self.filename}")

Functions

check_table(table_name: str) -> bool

Check if a table exists in the database.

Parameters:

Name Type Description Default
table_name str

Name of the table to check.

 required

Returns:

Type Description
bool

True if the table exists, False otherwise.
Source code in financelib/database/sqlite.py 
135
136
137
138
139
140
141
142
143
144
def check_table(self, table_name: str) -> bool:
    """Check if a table exists in the database.

    Args:
        table_name: Name of the table to check.

    Returns:
        True if the table exists, False otherwise.
    """
    return check_table_exist(self.cursor, table_name)

close() -> None

Close the database connection.

Source code in financelib/database/sqlite.py 
192
193
194
195
def close(self) -> None:
    """Close the database connection."""
    self.conn.close()
    logger.info(f"Database closed: {self.filename}")

execute(query: str, params: Optional[tuple] = None) -> None

Execute an arbitrary SQL query.

Parameters:

Name Type Description Default
query str

SQL query string.

 required
params Optional[tuple]

Optional query parameters.

 None
Source code in financelib/database/sqlite.py 
146
147
148
149
150
151
152
153
154
155
156
157
158
def execute(self, query: str, params: Optional[tuple] = None) -> None:
    """Execute an arbitrary SQL query.

    Args:
        query: SQL query string.
        params: Optional query parameters.
    """
    if params:
        self.cursor.execute(query, params)
    else:
        self.cursor.execute(query)
    if not self._in_transaction:
        self.conn.commit()

fetchall(query: str, params: Optional[tuple] = None) -> List[Tuple]

Execute a query and return all results.

Parameters:

Name Type Description Default
query str

SQL query string.

 required
params Optional[tuple]

Optional query parameters.

 None

Returns:

Type Description
List[Tuple]

List of result tuples.
Source code in financelib/database/sqlite.py 
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
def fetchall(self, query: str, params: Optional[tuple] = None) -> List[Tuple]:
    """Execute a query and return all results.

    Args:
        query: SQL query string.
        params: Optional query parameters.

    Returns:
        List of result tuples.
    """
    if params:
        self.cursor.execute(query, params)
    else:
        self.cursor.execute(query)
    return self.cursor.fetchall()

fetchone(query: str, params: Optional[tuple] = None) -> Optional[Tuple]

Execute a query and return the first result.

Parameters:

Name Type Description Default
query str

SQL query string.

 required
params Optional[tuple]

Optional query parameters.

 None

Returns:

Type Description
Optional[Tuple]

First result tuple or None.
Source code in financelib/database/sqlite.py 
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
def fetchone(self, query: str, params: Optional[tuple] = None) -> Optional[Tuple]:
    """Execute a query and return the first result.

    Args:
        query: SQL query string.
        params: Optional query parameters.

    Returns:
        First result tuple or None.
    """
    if params:
        self.cursor.execute(query, params)
    else:
        self.cursor.execute(query)
    return self.cursor.fetchone()

insert_article(news: NewsModel) -> None

Insert a single news article into the database.

Parameters:

Name Type Description Default
news NewsModel

NewsModel instance to insert.

 required
Source code in financelib/database/sqlite.py 
60
61
62
63
64
65
66
67
68
def insert_article(self, news: NewsModel) -> None:
    """Insert a single news article into the database.

    Args:
        news: NewsModel instance to insert.
    """
    query, params = make_insert_article_query(news)
    self.cursor.execute(query, params)
    self.conn.commit()

insert_articles(news_list: List[NewsModel]) -> None

Insert multiple news articles in a single transaction (batch).

Parameters:

Name Type Description Default
news_list List[NewsModel]

List of NewsModel instances to insert.

 required
Source code in financelib/database/sqlite.py 
70
71
72
73
74
75
76
77
78
79
80
81
def insert_articles(self, news_list: List[NewsModel]) -> None:
    """Insert multiple news articles in a single transaction (batch).

    Args:
        news_list: List of NewsModel instances to insert.
    """
    if not news_list:
        return
    query, _ = make_insert_article_query(news_list[0])
    all_params = [make_insert_article_query(n)[1] for n in news_list]
    self.cursor.executemany(query, all_params)
    self.conn.commit()

insert_stock_data(stock_data: StockDataModel) -> None

Insert stock data into the database.

Parameters:

Name Type Description Default
stock_data StockDataModel

StockDataModel instance to insert.

 required
Source code in financelib/database/sqlite.py 
125
126
127
128
129
130
131
132
133
def insert_stock_data(self, stock_data: StockDataModel) -> None:
    """Insert stock data into the database.

    Args:
        stock_data: StockDataModel instance to insert.
    """
    query, params = make_insert_stock_data_query(stock_data)
    self.cursor.execute(query, params)
    self.conn.commit()

insert_stock_data_batch(stock_list: List[StockDataModel]) -> None

Insert multiple stock data records in a single transaction.

Parameters:

Name Type Description Default
stock_list List[StockDataModel]

List of StockDataModel instances.

 required
Source code in financelib/database/sqlite.py 
83
84
85
86
87
88
89
90
91
92
93
94
def insert_stock_data_batch(self, stock_list: List[StockDataModel]) -> None:
    """Insert multiple stock data records in a single transaction.

    Args:
        stock_list: List of StockDataModel instances.
    """
    if not stock_list:
        return
    query, _ = make_insert_stock_data_query(stock_list[0])
    all_params = [make_insert_stock_data_query(s)[1] for s in stock_list]
    self.cursor.executemany(query, all_params)
    self.conn.commit()

transaction() -> _Transaction

Create a transaction context manager.

Inside a transaction, operations are atomic — they all commit together on success, or all roll back on error.

Example

with db.transaction(): ... db.execute("INSERT INTO news (title) VALUES (?)", ("A",)) ... db.execute("INSERT INTO news (title) VALUES (?)", ("B",)) ... # auto-commits on success, rollback on error

Source code in financelib/database/sqlite.py 
111
112
113
114
115
116
117
118
119
120
121
122
123
def transaction(self) -> "_Transaction":
    """Create a transaction context manager.

    Inside a transaction, operations are atomic — they all commit
    together on success, or all roll back on error.

    Example:
        >>> with db.transaction():
        ...     db.execute("INSERT INTO news (title) VALUES (?)", ("A",))
        ...     db.execute("INSERT INTO news (title) VALUES (?)", ("B",))
        ...     # auto-commits on success, rollback on error
    """
    return self._Transaction(self)

PostgreSQL

financelib.database.postgres.PostgreSQL(connection_string: Optional[str] = None)

PostgreSQL database wrapper for FinanceLib.

Provides the same API as the SQLite wrapper for easy migration. Stores news articles, stock data, trade history, and indicator snapshots.

Parameters:

Name Type Description Default
connection_string Optional[str]

PostgreSQL connection string. Format: postgresql://user:password@host:port/dbname If not provided, reads from DATABASE_URL environment variable.

 None
Example

db = PostgreSQL("postgresql://user:pass@localhost:5432/financelib") db.insert_stock_data(stock_model) db.close()

Or use environment variable

import os os.environ["DATABASE_URL"] = "postgresql://user:pass@localhost/financelib" with PostgreSQL() as db: ... db.insert_article(news_model)

Source code in financelib/database/postgres.py 
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
def __init__(self, connection_string: Optional[str] = None) -> None:
    try:
        import psycopg2
    except ImportError:
        raise ImportError(
            "psycopg2 is required for PostgreSQL support. "
            "Install it with: pip install financelib[db] or pip install psycopg2-binary"
        )

    dsn = connection_string or os.getenv("DATABASE_URL", "")
    if not dsn:
        raise ConfigurationError(
            "No PostgreSQL connection string provided. "
            "Pass it as argument or set DATABASE_URL environment variable."
        )

    try:
        self.conn = psycopg2.connect(dsn)
        self.conn.autocommit = False
        self.cursor = self.conn.cursor()
        self._initialize_tables()
        logger.info("PostgreSQL database connected")
    except Exception as e:
        raise DatabaseError(f"Failed to connect to PostgreSQL: {e}") from e

Functions

check_table(table_name: str) -> bool

Check if a table exists.

Parameters:

Name Type Description Default
table_name str

Table name to check.

 required

Returns:

Type Description
bool

True if the table exists.
Source code in financelib/database/postgres.py 
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
def check_table(self, table_name: str) -> bool:
    """Check if a table exists.

    Args:
        table_name: Table name to check.

    Returns:
        True if the table exists.
    """
    self.cursor.execute(
        "SELECT EXISTS(SELECT 1 FROM information_schema.tables WHERE table_name = %s)",
        (table_name,),
    )
    result = self.cursor.fetchone()
    return bool(result and result[0])

close() -> None

Close the database connection.

Source code in financelib/database/postgres.py 
327
328
329
330
331
def close(self) -> None:
    """Close the database connection."""
    self.cursor.close()
    self.conn.close()
    logger.info("PostgreSQL connection closed")

execute(query: str, params: Optional[tuple] = None) -> None

Execute an arbitrary SQL query.

Parameters:

Name Type Description Default
query str

SQL query (use %s placeholders).

 required
params Optional[tuple]

Query parameters.

 None
Source code in financelib/database/postgres.py 
291
292
293
294
295
296
297
298
299
def execute(self, query: str, params: Optional[tuple] = None) -> None:
    """Execute an arbitrary SQL query.

    Args:
        query: SQL query (use %s placeholders).
        params: Query parameters.
    """
    self.cursor.execute(query, params)
    self.conn.commit()

fetchall(query: str, params: Optional[tuple] = None) -> List[Tuple]

Execute a query and return all rows.

Parameters:

Name Type Description Default
query str

SQL query.

 required
params Optional[tuple]

Query parameters.

 None

Returns:

Type Description
List[Tuple]

List of result tuples.
Source code in financelib/database/postgres.py 
301
302
303
304
305
306
307
308
309
310
311
312
def fetchall(self, query: str, params: Optional[tuple] = None) -> List[Tuple]:
    """Execute a query and return all rows.

    Args:
        query: SQL query.
        params: Query parameters.

    Returns:
        List of result tuples.
    """
    self.cursor.execute(query, params)
    return self.cursor.fetchall()

fetchone(query: str, params: Optional[tuple] = None) -> Optional[Tuple]

Execute a query and return first row.

Parameters:

Name Type Description Default
query str

SQL query.

 required
params Optional[tuple]

Query parameters.

 None

Returns:

Type Description
Optional[Tuple]

First result tuple or None.
Source code in financelib/database/postgres.py 
314
315
316
317
318
319
320
321
322
323
324
325
def fetchone(self, query: str, params: Optional[tuple] = None) -> Optional[Tuple]:
    """Execute a query and return first row.

    Args:
        query: SQL query.
        params: Query parameters.

    Returns:
        First result tuple or None.
    """
    self.cursor.execute(query, params)
    return self.cursor.fetchone()

insert_article(news: NewsModel) -> None

Insert a news article.

Parameters:

Name Type Description Default
news NewsModel

NewsModel instance.

 required
Source code in financelib/database/postgres.py 
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
def insert_article(self, news: NewsModel) -> None:
    """Insert a news article.

    Args:
        news: NewsModel instance.
    """
    self.cursor.execute(
        """INSERT INTO news (title, content, published_at, author, category,
           source, article_url, article_thumbnail_url)
           VALUES (%s, %s, %s, %s, %s, %s, %s, %s)""",
        (
            news.title, news.content, news.published_at, news.author,
            news.category, news.source, news.article_url, news.article_thumbnail_url,
        ),
    )
    self.conn.commit()

insert_articles(news_list: List[NewsModel]) -> None

Insert multiple news articles in a batch.

Parameters:

Name Type Description Default
news_list List[NewsModel]

List of NewsModel instances.

 required
Source code in financelib/database/postgres.py 
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
def insert_articles(self, news_list: List[NewsModel]) -> None:
    """Insert multiple news articles in a batch.

    Args:
        news_list: List of NewsModel instances.
    """
    for news in news_list:
        self.cursor.execute(
            """INSERT INTO news (title, content, published_at, author, category,
               source, article_url, article_thumbnail_url)
               VALUES (%s, %s, %s, %s, %s, %s, %s, %s)""",
            (
                news.title, news.content, news.published_at, news.author,
                news.category, news.source, news.article_url, news.article_thumbnail_url,
            ),
        )
    self.conn.commit()

insert_indicator_snapshot(symbol: str, rsi: float = 0, macd: float = 0, macd_signal: float = 0, sma_20: float = 0, sma_50: float = 0, ema_20: float = 0, bb_upper: float = 0, bb_middle: float = 0, bb_lower: float = 0, atr: float = 0, adx: float = 0, stochastic_k: float = 0, stochastic_d: float = 0, williams_r: float = 0) -> None

Insert a snapshot of all indicator values for a symbol.

Parameters:

Name Type Description Default
symbol str

Stock/crypto symbol.

 required
rsi float

RSI value.

 0
macd float

MACD line value.

 0
macd_signal float

MACD signal line value.

 0
sma_20 float

20-period SMA.

 0
sma_50 float

50-period SMA.

 0
ema_20 float

20-period EMA.

 0
bb_upper float

Bollinger upper band.

 0
bb_middle float

Bollinger middle band.

 0
bb_lower float

Bollinger lower band.

 0
atr float

Average True Range.

 0
adx float

Average Directional Index.

 0
stochastic_k float

Stochastic %K.

 0
stochastic_d float

Stochastic %D.

 0
williams_r float

Williams %R.

 0
Source code in financelib/database/postgres.py 
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
def insert_indicator_snapshot(
    self,
    symbol: str,
    rsi: float = 0, macd: float = 0, macd_signal: float = 0,
    sma_20: float = 0, sma_50: float = 0, ema_20: float = 0,
    bb_upper: float = 0, bb_middle: float = 0, bb_lower: float = 0,
    atr: float = 0, adx: float = 0,
    stochastic_k: float = 0, stochastic_d: float = 0,
    williams_r: float = 0,
) -> None:
    """Insert a snapshot of all indicator values for a symbol.

    Args:
        symbol: Stock/crypto symbol.
        rsi: RSI value.
        macd: MACD line value.
        macd_signal: MACD signal line value.
        sma_20: 20-period SMA.
        sma_50: 50-period SMA.
        ema_20: 20-period EMA.
        bb_upper: Bollinger upper band.
        bb_middle: Bollinger middle band.
        bb_lower: Bollinger lower band.
        atr: Average True Range.
        adx: Average Directional Index.
        stochastic_k: Stochastic %K.
        stochastic_d: Stochastic %D.
        williams_r: Williams %R.
    """
    self.cursor.execute(
        """INSERT INTO indicator_snapshots
           (symbol, rsi, macd, macd_signal, sma_20, sma_50, ema_20,
            bb_upper, bb_middle, bb_lower, atr, adx,
            stochastic_k, stochastic_d, williams_r)
           VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)""",
        (symbol, rsi, macd, macd_signal, sma_20, sma_50, ema_20,
         bb_upper, bb_middle, bb_lower, atr, adx,
         stochastic_k, stochastic_d, williams_r),
    )
    self.conn.commit()

insert_stock_data(stock_data: StockDataModel) -> None

Insert stock data.

Parameters:

Name Type Description Default
stock_data StockDataModel

StockDataModel instance.

 required
Source code in financelib/database/postgres.py 
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
def insert_stock_data(self, stock_data: StockDataModel) -> None:
    """Insert stock data.

    Args:
        stock_data: StockDataModel instance.
    """
    self.cursor.execute(
        """INSERT INTO stock_data (symbol, price, open_price, high, low,
           volume, change, change_percent, currency, timestamp)
           VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s)""",
        (
            stock_data.symbol, stock_data.price, stock_data.open_price,
            stock_data.high, stock_data.low, stock_data.volume,
            stock_data.change, stock_data.change_percent,
            stock_data.currency, stock_data.timestamp,
        ),
    )
    self.conn.commit()

insert_trade(symbol: str, side: str, price: float, amount: float, market: str = 'spot', leverage: int = 1, pnl: float = 0.0, paper: bool = True) -> None

Insert a trade record.

Parameters:

Name Type Description Default
symbol str

Trading pair symbol.

 required
side str

Trade side ('buy', 'sell').

 required
price float

Execution price.

 required
amount float

Trade amount.

 required
market str

Market type.

 'spot'
leverage int

Leverage used.

 1
pnl float

Profit/loss.

 0.0
paper bool

Whether this was a paper trade.

 True
Source code in financelib/database/postgres.py 
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
def insert_trade(
    self,
    symbol: str,
    side: str,
    price: float,
    amount: float,
    market: str = "spot",
    leverage: int = 1,
    pnl: float = 0.0,
    paper: bool = True,
) -> None:
    """Insert a trade record.

    Args:
        symbol: Trading pair symbol.
        side: Trade side ('buy', 'sell').
        price: Execution price.
        amount: Trade amount.
        market: Market type.
        leverage: Leverage used.
        pnl: Profit/loss.
        paper: Whether this was a paper trade.
    """
    self.cursor.execute(
        """INSERT INTO trade_history (symbol, side, price, amount, market,
           leverage, pnl, paper) VALUES (%s, %s, %s, %s, %s, %s, %s, %s)""",
        (symbol, side, price, amount, market, leverage, pnl, paper),
    )
    self.conn.commit()

See also: Data Models for NewsModel, StockDataModel, TradeRecord.