tests.utils.DailyBarWriterFromDataFrames.check_uint_safe() - Code Metrics - Inspection of "Q 2.0" - quantopian/zipline - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Pull Request — master (#858)

by Eddie

created 2016-01-06 18:38 UTC

check_uint_safe() A

↳ Parent: tests.utils.DailyBarWriterFromDataFrames

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Metric	Value
cc	2
dl	0
loc	5
rs	9.4286

from numpy import (
    float64,
    uint32
)
from bcolz import ctable

from zipline.data.us_equity_pricing import (
    BcolzDailyBarWriter,
    OHLC,
    UINT32_MAX
)


class DailyBarWriterFromDataFrames(BcolzDailyBarWriter):

    _csv_dtypes = {
        'open': float64,
        'high': float64,
        'low': float64,
        'close': float64,
        'volume': float64,
    }

    def __init__(self, asset_map):
        self._asset_map = asset_map

    def gen_tables(self, assets):
        for asset in assets:
            yield asset, ctable.fromdataframe(assets[asset])

    def to_uint32(self, array, colname):
        arrmax = array.max()
        if colname in OHLC:
            self.check_uint_safe(arrmax * 1000, colname)
            return (array * 1000).astype(uint32)
        elif colname == 'volume':
            self.check_uint_safe(arrmax, colname)
            return array.astype(uint32)
        elif colname == 'day':
            nanos_per_second = (1000 * 1000 * 1000)
            self.check_uint_safe(arrmax.view(int) / nanos_per_second, colname)
            return (array.view(int) / nanos_per_second).astype(uint32)

    @staticmethod
    def check_uint_safe(value, colname):
        if value >= UINT32_MAX:
            raise ValueError(
                "Value %s from column '%s' is too large" % (value, colname)
            )


1			from numpy import (
2			float64,
3			uint32
4			)
5			from bcolz import ctable
6
7			from zipline.data.us_equity_pricing import (
8			BcolzDailyBarWriter,
9			OHLC,
10			UINT32_MAX
11			)
12
13
14			class DailyBarWriterFromDataFrames(BcolzDailyBarWriter):
			0 ignored issues – show Duplication introduced 2015-11-23 19:49 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Duplicated code is one of the most pungent code smells. If you need to duplicate the same code in three or more different places, we strongly encourage you to look into extracting the code into a single class or operation. You can also find more detailed suggestions in the “Code” section of your repository. Loading history...
15			_csv_dtypes = {
16			'open': float64,
17			'high': float64,
18			'low': float64,
19			'close': float64,
20			'volume': float64,
21			}
22
23			def __init__(self, asset_map):
24			self._asset_map = asset_map
25
26			def gen_tables(self, assets):
27			for asset in assets:
28			yield asset, ctable.fromdataframe(assets[asset])
29
30			def to_uint32(self, array, colname):
31			arrmax = array.max()
32			if colname in OHLC:
33			self.check_uint_safe(arrmax * 1000, colname)
34			return (array * 1000).astype(uint32)
35			elif colname == 'volume':
36			self.check_uint_safe(arrmax, colname)
37			return array.astype(uint32)
38			elif colname == 'day':
39			nanos_per_second = (1000 * 1000 * 1000)
40			self.check_uint_safe(arrmax.view(int) / nanos_per_second, colname)
41			return (array.view(int) / nanos_per_second).astype(uint32)
42
43			@staticmethod
44			def check_uint_safe(value, colname):
45			if value >= UINT32_MAX:
46			raise ValueError(
47			"Value %s from column '%s' is too large" % (value, colname)
48			)
49

quantopian / zipline

Pull Request — master (#858)

check_uint_safe() A

Complexity

Size

Duplication

Duplication Side-by-Side

Filter issues like