Creating Temporary Databases

Ticks can be used to create in-memory data for testing query, but sometimes it is necessary to create persistent data, which will be saved on the disk between the queries, data can be saved with different symbol and tick types.

Creating DB

Below the small example of how database can be created, filled with data and then read.

db = otp.DB("MY_DB")
db.add(otp.Tick(x=1), symbol="A", tick_type="TRD")
db.add(otp.Tick(y=2), symbol="B", tick_type="TRD")
session.use(db)
data = otp.Custom(db, symbol=["A", "B"], tick_type="TRD")
df = data.to_df()
print(df)

The result is:

        Time  X  Y
0 2003-12-01  1  0
1 2003-12-01  0  2

For adding data to the database you should use add() method.

Warning

Onetick has the following behavior worse to know:

• Use uppercase names only Mind the name of columns though the fields names were specified in lower case the columns names of read data are in uppercase, that is the expected behavior of Onetick, that’s why we suggest using uppercase names only.

• Do not try to rewrite the ticks If you will add the same ticks the old ones won’t be updated.

In order to read date from DB it should be added to the session by use() method, after it is possible to fetch data by using Custom object. You should specify the symbol and tick type of selected data, symbol can be passed as the collection of names or by Source object, see methods’ api reference and example below.

Note

In the example above the ticks has different schema, in such case the non existed field will be created and initialized by default value of the type. If several ticks has the same name for field of different types, it will cause the error.

Using otp.Custom as symbol

Any Source instance can be used as symbol parameter for fetching date from database, in the example below the temporary database will be created, it will have the list of all symbols, the query will extract the list filter symbol and then fetch the ticks of such symbols.

# create DB
db = otp.DB("MY_DB")
db.add(otp.Ticks(SYMBOL=["A", "AA", "B"]), symbol="S", tick_type="SYMBOLS")
db.add(otp.Tick(X=1), symbol="A", tick_type="TT")
db.add(otp.Tick(X=2), symbol="AA", tick_type="TT")
db.add(otp.Tick(X=3), symbol="B", tick_type="TT")

session.use(db)

# extract symbols and filter them
symbols = otp.Custom(db, symbol="S", tick_type="SYMBOLS")
symbols, _ = symbols[symbols["SYMBOL"].str.contains("A")]
symbols.rename(dict(SYMBOL="SYMBOL_NAME"), inplace=True)

# extract data with filtered symbols
data = otp.Custom(db, symbol=symbols, tick_type="TT")
df = data.to_df()
print(df)

        Time  X
0 2003-12-01  1
1 2003-12-01  2

Reuse Database

There is not need to create the same database in every test, they can be created only ones as class fixture, see the example below.

import pytest
import onetick.py as otp
from onetick.test.fixtures import c_session as session

class TestWithDB1:
    @pytest.fixture(scope="class")
    def db(self):
        db = otp.DB("MY_DB")
        db.add(otp.Tick(x=1), symbol="AA", tick_type="TRD")
        db.add(otp.Tick(x=2), symbol="BA", tick_type="TRD")
        yield db

    @pytest.fixture(scope="class")
    def session(self, c_session, db):
        c_session.use(db)
        yield c_session

    def test1(self, session, db):
        pass

    def test2(self, session, db):
        pass