Skip to content

Api strategies

A class for interacting with OneQuant strategies.

Source code in onequant/api/strategies.py
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
class OqStrategies:
    """A class for interacting with OneQuant strategies."""

    def __init__(self, wrapper=None):
        """Initializes an OqStrategies object.

        Args:
            wrapper (object, optional): An object containing the API and username. Defaults to None.
        """
        self.api = wrapper.api
        self.username = wrapper.username

    def _query(self, router, params=None):
        """Sends a GET request to the specified router with the given parameters.

        Args:
            router (str): The router to send the request to.
            params (dict, optional): The parameters to include in the request. Defaults to None.

        Returns:
            tuple: A tuple containing the data and code from the response.
        """
        result = self.api.request(method='get', router=router, params=params)
        if result['code'] != 200:
            raise Exception(f'An error occurred while retrieving tdegine data! code is {result["code"]}')
        return result['data']

    @_pd
    def _query_pd(self, params=None, router=None):
        """Sends a GET request and returns the data as a pandas DataFrame.

        Args:
            params (dict, optional): The parameters to include in the request. Defaults to None.
            router (str, optional): The router to send the request to. Defaults to None.

        Returns:
            pandas.DataFrame: The data from the response as a pandas DataFrame.
        """
        result = self.api.request(method='get', router=router, params=params)
        return result['data'], result['code']

    @_pd
    @_pagination
    def _query_pd_pg(self, params=None, router=None):
        """Sends a GET request and returns the data as a paginated pandas DataFrame.

        Args:
            params (dict, optional): The parameters to include in the request. Defaults to None.
            router (str, optional): The router to send the request to. Defaults to None.

        Returns:
            dict: A dictionary containing the paginated data and metadata from the response.
        """
        result = self.api.request(method='get', router=router, params=params)
        return result

    @_pd
    @tddata_2_list
    def _querytd_pd(self, router, params=None):
        """Sends a GET request and returns the data as a pandas DataFrame.

        Args:
            router (str): The router to send the request to.
            params (dict, optional): The parameters to include in the request. Defaults to None.

        Returns:
            pandas.DataFrame: The data from the response as a pandas DataFrame.
        """
        result = self.api.request(method='get', router=router, params=params)
        return result

    def strategy_base(self):
        """Returns the base information for all strategies.

        Returns:
            pandas.DataFrame: The base information for all strategies.
        """
        return self._query_pd(router='/strategy/info/base/query')

    def strategy_list(self, base_ea=None, base_tf=None, is_running=None, mark=None, min_tf=None):
        """Returns a paginated list of strategies that match the specified criteria.

        Args:
            base_ea (str, optional): The base EA to filter by. Defaults to None.
            base_tf (str, optional): The base time frame to filter by. Defaults to None.
            is_running (bool, optional): Whether the strategy is currently running. Defaults to None.
            mark (int, optional): The mark index to filter by. Defaults to None.

        Returns:
            dict: A dictionary containing the paginated list of strategies and metadata.
        """
        params = {
            'base_ea': base_ea,
            'base_tf': base_tf,
            'is_running': is_running,
            'mark_index': mark,
            'min_tf': min_tf,
        }
        return self._query_pd(router='/strategy/info/querypro', params=params)

    def strategy_report(
        self,
        strategy=None,
        base_ea=None,
        test_codes=None,
        base_tf=None,
        min_tf=None,
        min_netvalue=None,
        min_sharpe=None,
        min_annual_returns=None,
        min_calmar=None,
        min_sortino=None,
        max_margin=None,
        min_tradetimes=None,
        is_running=None,
    ):
        """Returns a paginated report of the specified strategy.

        Args:
            strategy (str, optional): The strategy to retrieve the report for. Defaults to None.
            base_ea (str, optional): The base EA to filter by. Defaults to None.
            test_codes (str, optional): The test codes to filter by. Defaults to None.
            base_tf (str, optional): The base time frame to filter by. Defaults to None.

        Returns:
            dict: A dictionary containing the paginated report and metadata.
        """
        params = {
            'strategy': strategy,
            'base_ea': base_ea,
            'base_tf': base_tf,
            'test_codes': test_codes,
            'min_tf': min_tf,
            'min_netvalue': min_netvalue,
            'min_sharpe': min_sharpe,
            'min_annual_returns': min_annual_returns,
            'min_calmar': min_calmar,
            'min_sortino': min_sortino,
            'max_margin': max_margin,
            'min_tradetimes': min_tradetimes,
            'status': -1 if is_running is None else (1 if is_running else 0),
        }
        return self._query_pd(router='/strategy/analyse/report/querypro', params=params)

    def strategy_netvalue(self, strategy_id=None):
        """Returns the net value for the specified strategy.

        Args:
            strategy_id (str, optional): The ID of the strategy to retrieve the net value for. Defaults to None.

        Returns:
            pandas.DataFrame: The net value for the specified strategy.
        """
        params = {'strategy_id': strategy_id}
        return self._querytd_pd(router='/strategy/analyse/netequity/query', params=params)

    def strategy_record(self, strategy_id=None):
        """Returns the record for the specified strategy.

        Args:
            strategy_id (str, optional): The ID of the strategy to retrieve the record for. Defaults to None.

        Returns:
            pandas.DataFrame: The record for the specified strategy.
        """
        params = {'strategy_id': strategy_id}
        return self._querytd_pd(router='/strateg/analyse/record/query', params=params)

__init__(wrapper=None)

Initializes an OqStrategies object.

Parameters:

Name Type Description Default
wrapper object

An object containing the API and username. Defaults to None.

None
Source code in onequant/api/strategies.py
 8
 9
10
11
12
13
14
15
def __init__(self, wrapper=None):
    """Initializes an OqStrategies object.

    Args:
        wrapper (object, optional): An object containing the API and username. Defaults to None.
    """
    self.api = wrapper.api
    self.username = wrapper.username

strategy_base()

Returns the base information for all strategies.

Returns:

Type Description

pandas.DataFrame: The base information for all strategies.

Source code in onequant/api/strategies.py
76
77
78
79
80
81
82
def strategy_base(self):
    """Returns the base information for all strategies.

    Returns:
        pandas.DataFrame: The base information for all strategies.
    """
    return self._query_pd(router='/strategy/info/base/query')

strategy_list(base_ea=None, base_tf=None, is_running=None, mark=None, min_tf=None)

Returns a paginated list of strategies that match the specified criteria.

Parameters:

Name Type Description Default
base_ea str

The base EA to filter by. Defaults to None.

None
base_tf str

The base time frame to filter by. Defaults to None.

None
is_running bool

Whether the strategy is currently running. Defaults to None.

None
mark int

The mark index to filter by. Defaults to None.

None

Returns:

Name Type Description
dict

A dictionary containing the paginated list of strategies and metadata.

Source code in onequant/api/strategies.py
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
def strategy_list(self, base_ea=None, base_tf=None, is_running=None, mark=None, min_tf=None):
    """Returns a paginated list of strategies that match the specified criteria.

    Args:
        base_ea (str, optional): The base EA to filter by. Defaults to None.
        base_tf (str, optional): The base time frame to filter by. Defaults to None.
        is_running (bool, optional): Whether the strategy is currently running. Defaults to None.
        mark (int, optional): The mark index to filter by. Defaults to None.

    Returns:
        dict: A dictionary containing the paginated list of strategies and metadata.
    """
    params = {
        'base_ea': base_ea,
        'base_tf': base_tf,
        'is_running': is_running,
        'mark_index': mark,
        'min_tf': min_tf,
    }
    return self._query_pd(router='/strategy/info/querypro', params=params)

strategy_netvalue(strategy_id=None)

Returns the net value for the specified strategy.

Parameters:

Name Type Description Default
strategy_id str

The ID of the strategy to retrieve the net value for. Defaults to None.

None

Returns:

Type Description

pandas.DataFrame: The net value for the specified strategy.

Source code in onequant/api/strategies.py
149
150
151
152
153
154
155
156
157
158
159
def strategy_netvalue(self, strategy_id=None):
    """Returns the net value for the specified strategy.

    Args:
        strategy_id (str, optional): The ID of the strategy to retrieve the net value for. Defaults to None.

    Returns:
        pandas.DataFrame: The net value for the specified strategy.
    """
    params = {'strategy_id': strategy_id}
    return self._querytd_pd(router='/strategy/analyse/netequity/query', params=params)

strategy_record(strategy_id=None)

Returns the record for the specified strategy.

Parameters:

Name Type Description Default
strategy_id str

The ID of the strategy to retrieve the record for. Defaults to None.

None

Returns:

Type Description

pandas.DataFrame: The record for the specified strategy.

Source code in onequant/api/strategies.py
161
162
163
164
165
166
167
168
169
170
171
def strategy_record(self, strategy_id=None):
    """Returns the record for the specified strategy.

    Args:
        strategy_id (str, optional): The ID of the strategy to retrieve the record for. Defaults to None.

    Returns:
        pandas.DataFrame: The record for the specified strategy.
    """
    params = {'strategy_id': strategy_id}
    return self._querytd_pd(router='/strateg/analyse/record/query', params=params)

strategy_report(strategy=None, base_ea=None, test_codes=None, base_tf=None, min_tf=None, min_netvalue=None, min_sharpe=None, min_annual_returns=None, min_calmar=None, min_sortino=None, max_margin=None, min_tradetimes=None, is_running=None)

Returns a paginated report of the specified strategy.

Parameters:

Name Type Description Default
strategy str

The strategy to retrieve the report for. Defaults to None.

None
base_ea str

The base EA to filter by. Defaults to None.

None
test_codes str

The test codes to filter by. Defaults to None.

None
base_tf str

The base time frame to filter by. Defaults to None.

None

Returns:

Name Type Description
dict

A dictionary containing the paginated report and metadata.

Source code in onequant/api/strategies.py
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
def strategy_report(
    self,
    strategy=None,
    base_ea=None,
    test_codes=None,
    base_tf=None,
    min_tf=None,
    min_netvalue=None,
    min_sharpe=None,
    min_annual_returns=None,
    min_calmar=None,
    min_sortino=None,
    max_margin=None,
    min_tradetimes=None,
    is_running=None,
):
    """Returns a paginated report of the specified strategy.

    Args:
        strategy (str, optional): The strategy to retrieve the report for. Defaults to None.
        base_ea (str, optional): The base EA to filter by. Defaults to None.
        test_codes (str, optional): The test codes to filter by. Defaults to None.
        base_tf (str, optional): The base time frame to filter by. Defaults to None.

    Returns:
        dict: A dictionary containing the paginated report and metadata.
    """
    params = {
        'strategy': strategy,
        'base_ea': base_ea,
        'base_tf': base_tf,
        'test_codes': test_codes,
        'min_tf': min_tf,
        'min_netvalue': min_netvalue,
        'min_sharpe': min_sharpe,
        'min_annual_returns': min_annual_returns,
        'min_calmar': min_calmar,
        'min_sortino': min_sortino,
        'max_margin': max_margin,
        'min_tradetimes': min_tradetimes,
        'status': -1 if is_running is None else (1 if is_running else 0),
    }
    return self._query_pd(router='/strategy/analyse/report/querypro', params=params)