Source code for onetick.py.core._source.symbol
import warnings
from onetick.py.core._source._symbol_param_column import _SymbolParamColumn
import onetick.py.types as ott
globals().setdefault("__warningregistry__", {}) # otherwise doctests fails
[docs]class SymbolType:
def __init__(self):
"""
You can get symbol name and symbol parameters with this class.
Examples
--------
>>> symbols = otp.Symbols('SOME_DB')
>>> symbols['PARAM'] = 'PAM'
>>> ticks = otp.DataSource('SOME_DB', tick_type='TT')
>>> ticks['SYMBOL_PARAM'] = ticks.Symbol.PARAM
>>> ticks['SYMBOL_NAME'] = ticks.Symbol.name
>>> ticks = otp.merge([ticks], symbols=symbols)
>>> otp.run(ticks)
Time X SYMBOL_PARAM SYMBOL_NAME
0 2003-12-01 00:00:00.000 1 PAM S1
1 2003-12-01 00:00:00.000 -3 PAM S2
2 2003-12-01 00:00:00.001 2 PAM S1
3 2003-12-01 00:00:00.001 -2 PAM S2
4 2003-12-01 00:00:00.002 3 PAM S1
5 2003-12-01 00:00:00.002 -1 PAM S2
See also
--------
| :ref:`Databases, symbols, and tick types <symbols_concept>`
| :ref:`Symbol Parameters Objects`
"""
self._name = _SymbolParamColumn("_SYMBOL_NAME", str)
@property
def name(self):
"""
Get symbol name.
Returns
-------
_SymbolParamColumn
Examples
--------
>>> symbols = otp.Symbols('SOME_DB')
>>> ticks = otp.DataSource('SOME_DB', tick_type='TT')
>>> ticks['SYMBOL_NAME'] = ticks.Symbol.name
>>> ticks = otp.merge([ticks], symbols=symbols)
>>> otp.run(ticks)
Time X SYMBOL_NAME
0 2003-12-01 00:00:00.000 1 S1
1 2003-12-01 00:00:00.000 -3 S2
2 2003-12-01 00:00:00.001 2 S1
3 2003-12-01 00:00:00.001 -2 S2
4 2003-12-01 00:00:00.002 3 S1
5 2003-12-01 00:00:00.002 -1 S2
"""
return self._name
@staticmethod
def _get(item, dtype=str):
column = _SymbolParamColumn(f"_SYMBOL_PARAM.{item}", str)
unit = None
if dtype is ott.nsectime:
unit = 'ns'
if dtype is ott.msectime:
unit = 'ms'
if unit is not None:
return column.str.to_datetime(unit=unit)
return column.astype(dtype)
[docs] def __getattr__(self, item):
"""
Get symbol parameter by name. Notice, that symbol parameter type will be string.
.. deprecated:: 1.74.0
Please, use :py:meth:`~onetick.py.core._source.symbol.__getitem__` method.
Returns
-------
_SymbolParamColumn
Examples
--------
>>> symbols = otp.Symbols('SOME_DB')
>>> symbols['PARAM'] = 'PAM'
>>> ticks = otp.DataSource('SOME_DB', tick_type='TT')
>>> ticks['SYMBOL_PARAM'] = ticks.Symbol.PARAM
>>> ticks = otp.merge([ticks], symbols=symbols)
>>> otp.run(ticks)
Time X SYMBOL_PARAM
0 2003-12-01 00:00:00.000 1 PAM
1 2003-12-01 00:00:00.000 -3 PAM
2 2003-12-01 00:00:00.001 2 PAM
3 2003-12-01 00:00:00.001 -2 PAM
4 2003-12-01 00:00:00.002 3 PAM
5 2003-12-01 00:00:00.002 -1 PAM
"""
warnings.warn("`__getattr__` method is deprecated. Please, use `__getitem__` method instead.",
DeprecationWarning)
return self._get(item)
[docs] def __getitem__(self, item):
"""
Get symbol parameter by name.
Parameters
----------
item: tuple
The first parameter is string - symbol parameter name. The second one is symbol parameter type.
Returns
-------
_SymbolParamColumn
Examples
--------
The second parameter is symbol parameter's type.
>>> symbols = otp.Symbols('SOME_DB')
>>> symbols['PARAM'] = 5
>>> ticks = otp.DataSource('SOME_DB', tick_type='TT')
>>> ticks['SYMBOL_PARAM'] = ticks.Symbol['PARAM', int] + 1
>>> ticks['SYMBOL_PARAM'].dtype
<class 'int'>
>>> ticks = otp.merge([ticks], symbols=symbols)
>>> otp.run(ticks)
Time X SYMBOL_PARAM
0 2003-12-01 00:00:00.000 1 6
1 2003-12-01 00:00:00.000 -3 6
2 2003-12-01 00:00:00.001 2 6
3 2003-12-01 00:00:00.001 -2 6
4 2003-12-01 00:00:00.002 3 6
5 2003-12-01 00:00:00.002 -1 6
It also works with :py:class:`~onetick.py.msectime` and :py:class:`~onetick.py.nsectime` types:
>>> symbols = otp.Symbols('SOME_DB')
>>> symbols['NSECTIME_PARAM'] = symbols['Time'] + otp.Nano(100)
>>> symbols['MSECTIME_PARAM'] = symbols['Time'] + otp.Milli(1)
>>> ticks = otp.DataSource('SOME_DB', tick_type='TT')
>>> ticks['NSECTIME_PARAM'] = ticks.Symbol['NSECTIME_PARAM', otp.nsectime] + otp.Nano(1)
>>> ticks['MSECTIME_PARAM'] = ticks.Symbol['MSECTIME_PARAM', otp.msectime] + otp.Milli(1)
>>> ticks['NSECTIME_PARAM'].dtype
<class 'onetick.py.types.nsectime'>
>>> ticks['MSECTIME_PARAM'].dtype
<class 'onetick.py.types.msectime'>
>>> ticks = otp.merge([ticks], symbols=symbols)
>>> otp.run(ticks)
Time X NSECTIME_PARAM MSECTIME_PARAM
0 2003-12-01 00:00:00.000 1 2003-12-01 00:00:00.000000101 2003-12-01 00:00:00.002
1 2003-12-01 00:00:00.000 -3 2003-12-01 00:00:00.000000101 2003-12-01 00:00:00.002
2 2003-12-01 00:00:00.001 2 2003-12-01 00:00:00.000000101 2003-12-01 00:00:00.002
3 2003-12-01 00:00:00.001 -2 2003-12-01 00:00:00.000000101 2003-12-01 00:00:00.002
4 2003-12-01 00:00:00.002 3 2003-12-01 00:00:00.000000101 2003-12-01 00:00:00.002
5 2003-12-01 00:00:00.002 -1 2003-12-01 00:00:00.000000101 2003-12-01 00:00:00.002
"""
if not isinstance(item, tuple):
raise ValueError(f"tuple should be passed, but {type(item)} was passed")
if len(item) != 2:
raise ValueError(f"tuple's length should be 2 but it is {len(item)}")
item, dtype = item
return self._get(item, dtype)
Symbol = SymbolType() # noqa mypy fix