quantlab.acquisition.wrds.taq.WrdsSession#

class quantlab.acquisition.wrds.taq.WrdsSession(username)[source]#

Bases: object

A pool of read-only PostgreSQL connections to WRDS, opened on first use.

Constructing a session makes no network call. A query takes an idle connection from the pool or opens a new one, up to MAX_CONNECTIONS; when every connection is busy the query waits for one. Sequential use therefore opens exactly one connection, and max_workers download threads open at most max_workers. Get the session through shared() rather than constructing one per batch, so the whole run counts against one pool: a WRDS account may hold only a few connections at once. Once the driver reports an error, or a connect fails, the session is broken for the rest of the run and never reconnects; the next run resumes from the pages on disk.

The host, port and database are class constants passed directly to psycopg2.connect, never read from a config or the environment, so no config file can send the credentials to another server. This code never handles the password: libpq reads it from ~/.pgpass (or the file named by $PGPASSFILE), and the session checks that the file has a matching entry before it tries to connect.

The methods fall into three groups: pure SQL builders that need no connection (table_identifier, where_clause, copy_query, count_query), general query helpers used by every WRDS product (schema_usable, fetch_rows, copy_csv), and TAQ-specific methods built on those helpers.

Parameters:

usernamestr: The WRDS username, normally read from WRDS_USERNAME by shared().

Examples

Needs WRDS_USERNAME and a matching ~/.pgpass line; the first query opens the first connection:

session = WrdsSession.shared()
days = session.trading_days(2024)
WrdsSession.close_shared()

COPY_SPOOL_BYTES = 268435456#: Up to this many bytes of one COPY result stay in memory before the buffer moves to a temporary file.

MAX_CONNECTIONS = 6#: Connections one session opens at most. A WRDS account may hold about seven connections at once; one is left free for another client.

PGPASS_LINE_HINT = 'wrds-pgdata.wharton.upenn.edu:9737:wrds:$WRDS_USERNAME:<password>'#: The .pgpass line quoted in error messages. <password> is a placeholder; this code never reads the real password.

REFUSED_ENV = ('PGHOSTADDR', 'PGSERVICE', 'PGSERVICEFILE')#: libpq environment variables that could redirect the connection away from HOST: PGHOSTADDR overrides the address the host name resolves to, and a service file can supply host, port and user. If any is set, the session refuses to connect. PGHOST is harmless because the host is always passed explicitly.

assert_entitled(years)[source]#

Check that the account may read the TAQ schema of every year given.

Called before the first data query of a download, so a year outside the subscription stops the run before any data is copied, instead of failing every batch of every day.

Parameters:

yearsiterable of int: Calendar years.

Raises:

WrdsEntitlementError: Naming every taqm_YYYY schema the account cannot read.

Examples

Needs a live session:

session.assert_entitled(range(2020, 2025))

close()[source]#

Close every idle connection and refuse further queries.

A connection still in use by another thread is closed when that thread returns it.

Examples

>>> WrdsSession("someone").close()

classmethod close_shared()[source]#

Close and forget every shared session.

Examples

>>> WrdsSession.close_shared()

copy_csv(query)[source]#

Run a COPY ... TO STDOUT statement and return the CSV bytes.

The statement is passed to copy_expert unchanged, so no SQL text is assembled outside psycopg2.sql. Up to COPY_SPOOL_BYTES stay in memory before the buffer moves to a temporary file.

Parameters:

querypsycopg2.sql.Composable: A COPY (...) TO STDOUT statement.

Returns:

bytes: The CSV output.

Examples

Needs a live session:

raw = session.copy_csv(
    WrdsSession.copy_query(day, [("AAPL", None)], columns)
)

copy_nbbo_bars_csv(query)[source]#

Run a server-side bar resampling statement and return the CSV bytes.

Parameters:

queryquantlab.acquisition.wrds.nbbo_bars_sql.NbboBarsQuery: One day table and one symbol batch; its statement() is run unchanged.

Returns:

bytes: The CSV output, starting with a header line.

Examples

Needs a live session:

raw = session.copy_nbbo_bars_csv(query)

copy_nbbo_csv(day, pairs, columns)[source]#

Run copy_query for one day and batch and return the CSV bytes.

The statement is exactly the one copy_query builds, with no ordering or time condition added.

Parameters:

daydatetime.date: The trading day.
pairsiterable of tuple of (str, str or None): (root, suffix) pairs, as for where_clause.
columnsiterable of str: Column names to select, in this order.

Returns:

bytes: The CSV output, starting with a header line.

Examples

Needs a live session:

day = date(2024, 1, 24)
raw = session.copy_nbbo_csv(
    day, [("AAPL", None)], session.table_columns(day)
)

classmethod copy_query(day, pairs, columns)[source]#

Build the COPY statement for one day table and one symbol batch.

COPY ... TO STDOUT streams the result as CSV, which is much faster than fetching rows one by one. The statement has no ORDER BY, GROUP BY or DISTINCT and no time condition. The whole day is collected, and before 2018 the order in which the server stores the rows is the only way to order records that share a timestamp.

Parameters:

daydatetime.date: The trading day.
pairsiterable of tuple of (str, str or None): (root, suffix) pairs, as for where_clause.
columnsiterable of str: Column names to select, in this order.

Returns:

psycopg2.sql.Composed: The complete statement. The CSV output includes a header line.

Examples

>>> query = WrdsSession.copy_query(
...     date(2024, 1, 24), [("AAPL", None)], ("date", "time_m", "best_bid")
... )

which renders as:

COPY (SELECT "date", "time_m", "best_bid"
      FROM "taqm_2024"."complete_nbbo_20240124"
      WHERE sym_root = ANY(ARRAY['AAPL']) AND
            (sym_root, coalesce(sym_suffix, '')) IN (('AAPL', '')))
TO STDOUT WITH (FORMAT csv, HEADER true)

classmethod count_query(day, pairs)[source]#

Build SELECT count(*) over the same WHERE as copy_query.

Using the same condition guarantees that a count and the download it checks select the same rows.

Parameters:

daydatetime.date: The trading day.
pairsiterable of tuple of (str, str or None): (root, suffix) pairs, as for where_clause.

Returns:

psycopg2.sql.Composed: The count statement.

Examples

>>> query = WrdsSession.count_query(date(2024, 1, 24), [("AAPL", None)])

which renders as:

SELECT count(*) FROM "taqm_2024"."complete_nbbo_20240124"
WHERE sym_root = ANY(ARRAY['AAPL']) AND
      (sym_root, coalesce(sym_suffix, '')) IN (('AAPL', ''))

count_rows(day, pairs)[source]#

Return how many rows copy_query(day, pairs, ...) would return.

Parameters:

daydatetime.date: The trading day.
pairsiterable of tuple of (str, str or None): (root, suffix) pairs, as for where_clause.

Returns:

int: The row count.

Examples

Needs a live session:

session.count_rows(date(2024, 1, 24), [("AAPL", None)])

fetch_rows(query)[source]#

Execute query and return every row.

Meant for small catalogue and metadata reads. A data download uses copy_csv instead, which streams a full day of records without building a Python tuple per row.

Parameters:

querypsycopg2.sql.Composable or str: The statement to run.

Returns:

list of tuple: All result rows.

Examples

Needs a live session:

from psycopg2 import sql
rows = session.fetch_rows(sql.SQL("SELECT 1"))

has_schema_usage(year)[source]#

Return whether this account may read the TAQ schema taqm_{year}.

Parameters:

yearint: The calendar year.

Returns:

bool: The answer of schema_usable for that schema.

Examples

Needs a live session:

entitled = session.has_schema_usage(2024)

schema_usable(schema)[source]#

Return whether this account may read schema.

Reading a schema requires the USAGE privilege. The question is asked through the pg_namespace catalogue rather than by schema name, because asking by name raises for a schema that does not exist, and a driver error would break the session. A missing schema returns no row and reads as False, the same answer as “not subscribed”.

Parameters:

schemastr: The schema name, passed as a query parameter.

Returns:

bool: True if the account has USAGE on the schema.

Examples

Needs a live session:

usable = session.schema_usable("crsp_a_stock")

classmethod shared()[source]#

Return the process-wide session for the current WRDS_USERNAME.

The variable is read on every call, so a changed username gets its own session rather than the old one.

Returns:

WrdsSession: The same object on every call with the same username.

Raises:

RuntimeError: If WRDS_USERNAME is unset or empty.

Examples

With WRDS_USERNAME set (no connection is opened):

session = WrdsSession.shared()
session is WrdsSession.shared()
# True

table_columns(day)[source]#

Return the day table’s column names in the server’s column order.

Parameters:

daydatetime.date: The trading day.

Returns:

tuple of str: Column names, sorted by their position in the table.

Examples

Needs a live session:

columns = session.table_columns(date(2024, 1, 24))

static table_identifier(day)[source]#

Return the taqm_{YYYY}.complete_nbbo_{YYYYMMDD} table name for a day.

Parameters:

daydatetime.date: The trading day.

Returns:

psycopg2.sql.Identifier: The schema-qualified table name, safely quoted.

Examples

>>> from datetime import date
>>> WrdsSession.table_identifier(date(2024, 1, 24))
Identifier('taqm_2024', 'complete_nbbo_20240124')

trading_days(year)[source]#

Return every day that has a complete_nbbo table in taqm_{year}.

The days are read from the database catalogue (information_schema) rather than tested one calendar day at a time, so a missing table cannot be mistaken for a holiday.

Parameters:

yearint: The calendar year.

Returns:

list of datetime.date: The trading days, ascending.

Examples

Needs a live session:

days = session.trading_days(2024)

static where_clause(pairs)[source]#

Build the WHERE condition selecting exact (sym_root, sym_suffix) pairs.

TAQ splits a symbol into a root and a share-class suffix, so BRK.B is root BRK with suffix B. The first part of the condition, sym_root = ANY(roots), lets the server skip storage blocks that hold none of the roots; the second part picks the exact share classes. Every value is a sql.Literal, never text pasted into the SQL.

Parameters:

pairsiterable of tuple of (str, str or None): (root, suffix) pairs; a None or empty suffix means the plain root.

Returns:

psycopg2.sql.Composed: The condition, without the WHERE keyword.

Raises:

ValueError: If pairs is empty. Without a sym_root condition the query would scan a whole day table, which this class never does.

Examples

>>> clause = WrdsSession.where_clause([("AAPL", None), ("BRK", "B")])

which renders as:

sym_root = ANY(ARRAY['AAPL', 'BRK']) AND
(sym_root, coalesce(sym_suffix, '')) IN (('AAPL', ''), ('BRK', 'B'))