Pinform: An InfluxDB ORM (OTSM) for Python

PInfORM (Python InfluxDB ORM) is an Object/TimeSeries Mapping layer for connecting to InfluxDB in python.

Tested with:

• Python 3.6+
• InfluxDB 1.7.3 - 1.7.9

Use the following command to install using pip:

pip install pinform

Usage example

Create Measurement Models

First, create your measurement model in

from pinform import Measurement
from pinform.fields import FloatField
from pinform.tags import Tag

class OHLC(Measurement):
  class Meta:
    measurement_name = 'ohlc'

  symbol = Tag(null=False)
  open = FloatField(null=False)
  high = FloatField(null=False)
  low = FloatField(null=False)
  close = FloatField(null=False)

Create Measurement Instance

First, create your measurement model in

today_ohlc = OHLC(time_point=datetime.datetime.now(), symbol='AAPL', open=80.2, high=86.0, low=78.9, close=81.25)

Create InfluxClient

Then you must create an instance of InfluxClient to connect to database:

from pinform.client import InfluxClient

cli = InfluxClient(host="localhost", port=8086, database_name="defaultdb")

If the database needs authentication, use:

cli = InfluxClient(host="localhost", port=8086, database_name="defaultdb", username='your db username', password='your db password')

Save and Retrieve Points

To save data in database, use save_points or save_dataframe functions of InfluxClient:

ohlc = OHLC(time_point=datetime.datetime.now(), symbol='AAPL', open=100.6, high=102.5, low=90.4, close=94.2)
cli.save_points([ohlc])

To retrieve data from database, use load_points or load_points_as_dataframe functions of InfluxClient:

ohlc_points = cli.load_points(OHLC, {'symbol':'AAPL'})

Get Distinct Tag Values

To get distinct tag values from all measurements, use get_distinct_existing_tag_values function from InfluxClient:

tag_values = cli.get_distinct_existing_tag_values('symbol')

To get distinct tag values from an specific measurements,pass measurement to the previous function:

tag_values = cli.get_distinct_existing_tag_values('symbol', measurement=OHLC)

Fields

It's possible to use IntegerField, FloatField, BooleanField and StringField to save field values in InfluxDB. There are four other types of fields which help with storing fields with specific integer or string values. To create a field with multiple choice integer values, use MultipleChoiceIntegerField or EnumIntegerField classes. To create a field with multiple choice string values, use MultipleChoiceStringField or EnumStringField classes.

Example for MultipleChoiceStringField:

from pinform.fields import MultipleChoiceStringField

class WeatherInfo(Measurement):
  class Meta:
    measurement_name = 'weather_info'
  
  condition = MultipleChoiceStringField(options=['sunny','cloudy','rainy'], null=False)

Example for EnumStringField:

from enum import Enum
from pinform.fields import EnumStringField

class WeatherCondition(Enum):
  SUNNY = 'sunny'
  CLOUDY = 'cloudy'
  RAINY = 'rainy'


class WeatherInfo(Measurement):
  class Meta:
    measurement_name = 'weather_info'
  
  condition = EnumStringField(enum=WeatherCondition, null=False)

Advanced usage

Dynamic measurement names

It is possible to use 'MeasurementNameComponent's in measurement name wrapped in parenthesis, their value is replaced in the measurement name at runtime.

from pinform import MeasurementNameComponent

class OHLC(Measurement):
  class Meta:
    measurement_name = 'ohlc_(symbol)'
  
  symbol = MeasurementNameComponent(name='symbol')
  ...

Query Field and Pandas Series

Use get_fields_as_series function from InfluxClient to get fields of specific measurement class as Pandas Series. It's also possible to aggregate data and group by time. This function returnes a dict with aggregated field names as keys and pandas series as values.

from pinform.client import AggregationMode

series_dict = cli.get_fields_as_series(OHLC, 
                field_aggregations={'close': [AggregationMode.MEAN, AggregationMode.STDDEV]},
                tags={'symbol': 'AAPL'},
                time_range=(start_datetime, end_datetime),
                group_by_time_interval='10d')
mean_close_series = series_dict['mean_close']
stddev_close_series = series_dict['stddev_close']