我们从Python开源项目中，提取了以下50个代码示例，用于说明如何使用pytz.FixedOffset()。
def test_instance_timezone_aware_datetime_pytz_offset(self): # Impossible timezone of +21 (won't accidentally match the local offset) fixed_offset = pytz.FixedOffset(21 * 60) now = Pendulum.instance( datetime.now(fixed_offset) ) self.assertEqual(21, now.offset_hours) now = Pendulum.instance( datetime.now(), fixed_offset ) self.assertEqual(21, now.offset_hours) now = Pendulum.instance( datetime.now(fixed_offset), pytz.timezone('Europe/Paris') ) self.assertEqual(21, now.offset_hours)
def update_position(self): for _, pos in self.__shares.items(): inst = Instrument.objects.filter( exchange=pos['ExchangeID'], product_code=re.findall('[A-Za-z]+', pos['InstrumentID'])[0]).first() tick = json.loads(self.redis_client.get(pos['InstrumentID'])) if pos['Direction'] == ApiStruct.D_Buy: profit = Decimal(tick['LastPrice']) - Decimal(pos['OpenPrice']) else: profit = Decimal(pos['OpenPrice']) - Decimal(tick['LastPrice']) profit = profit * Decimal(pos['Volume']) * inst.volume_multiple Trade.objects.update_or_create( broker=self.__broker, strategy=self.__strategy, instrument=inst, code=pos['InstrumentID'], direction=DirectionType.LONG if pos['Direction'] == ApiStruct.D_Buy else DirectionType.SHORT, close_time__isnull=True, defaults={ 'open_time': datetime.datetime.strptime( pos['OpenDate']+'09', '%Y%m%d%H').replace(tzinfo=pytz.FixedOffset(480)), 'shares': pos['Volume'], 'filled_shares': pos['Volume'], 'avg_entry_price': Decimal(pos['OpenPrice']), 'cost': pos['Volume'] * Decimal(pos['OpenPrice']) * inst.fee_money * inst.volume_multiple + pos['Volume'] * inst.fee_volume, 'profit': profit, 'frozen_margin': Decimal(pos['Margin'])})
def start(self): self.redis_client.set('HEARTBEAT:TRADER', 1, ex=61) await self.query('TradingAccount') self.__shares.clear() await self.query('InvestorPositionDetail') # await self.collect_tick_stop() # await self.collect_quote() # day = datetime.datetime.strptime('20161031', '%Y%m%d').replace(tzinfo=pytz.FixedOffset(480)) # for inst in self.__strategy.instruments.all(): # # self.calc_signal(inst, day) # self.process_signal(inst) # order_list = await self.query('Order') # if order_list: # for order in order_list: # if order['OrderStatus'] == ApiStruct.OST_NotTouched and \ # order['OrderSubmitStatus'] == ApiStruct.OSS_InsertSubmitted: # self.__activeOrders[order['OrderRef']] = order # logger.info("?????: %s", self.__activeOrders) # inst_set = list() # for inst in Instrument.objects.all(): # inst_set += inst.all_inst.split(',') # await self.SubscribeMarketData(inst_set)
def fetch_bar(): day = datetime.datetime.strptime('20100416', '%Y%m%d').replace(tzinfo=pytz.FixedOffset(480)) end = datetime.datetime.strptime('20160118', '%Y%m%d').replace(tzinfo=pytz.FixedOffset(480)) tasks = [] while day <= end: tasks.append(is_trading_day(day)) day += datetime.timedelta(days=1) trading_days = [] for f in tqdm(asyncio.as_completed(tasks), total=len(tasks)): rst = await f trading_days.append(rst) tasks.clear() for day, trading in trading_days: if trading: tasks += [ asyncio.ensure_future(update_from_shfe(day)), asyncio.ensure_future(update_from_dce(day)), asyncio.ensure_future(update_from_czce(day)), asyncio.ensure_future(update_from_cffex(day)), ] print('task len=', len(tasks)) for f in tqdm(asyncio.as_completed(tasks), total=len(tasks)): await f
def fetch_bar2(): day = datetime.datetime.strptime('20160114', '%Y%m%d').replace(tzinfo=pytz.FixedOffset(480)) end = datetime.datetime.strptime('20160914', '%Y%m%d').replace(tzinfo=pytz.FixedOffset(480)) while day <= end: day, trading = await is_trading_day(day) if trading: print('process ', day) tasks = [ asyncio.ensure_future(update_from_shfe(day)), asyncio.ensure_future(update_from_dce(day)), asyncio.ensure_future(update_from_czce(day)), asyncio.ensure_future(update_from_cffex(day)), ] await asyncio.wait(tasks) day += datetime.timedelta(days=1) print('all done!') # asyncio.get_event_loop().run_until_complete(fetch_bar2()) # create_main_all() # fetch_from_quandl_all() # clean_dailybar() # load_kt_data()
def clean_daily_bar(): day = datetime.datetime.strptime('20100416', '%Y%m%d').replace(tzinfo=pytz.FixedOffset(480)) end = datetime.datetime.strptime('20160118', '%Y%m%d').replace(tzinfo=pytz.FixedOffset(480)) tasks = [] while day <= end: tasks.append(is_trading_day(day)) day += datetime.timedelta(days=1) trading_days = [] for f in tqdm(asyncio.as_completed(tasks), total=len(tasks)): rst = await f trading_days.append(rst) tasks.clear() for day, trading in trading_days: if not trading: DailyBar.objects.filter(time=day.date()).delete() print('done!')
def _get_localtime(send_job): timezone = pytz.FixedOffset(-send_job.utc_offset) time = pytz.utc.localize(send_job.scheduled_at) localtime = time.astimezone(timezone) return localtime
def __reduce__(self): return FixedOffset, (self._minutes, )
def __repr__(self): return 'pytz.FixedOffset(%d)' % self._minutes
def __reduce__(self): return FixedOffset, (self._minutes,)
def asn1_generalizedtime_to_dt(timestamp): """Convert ASN.1 GENERALIZEDTIME to datetime. Useful for deserialization of `OpenSSL.crypto.X509.get_notAfter` and `OpenSSL.crypto.X509.get_notAfter` outputs. """ dt = datetime.datetime.strptime(timestamp[:12], '%Y%m%d%H%M%S') if timestamp.endswith('Z'): tzinfo = pytz.utc else: sign = -1 if timestamp[-5] == '-' else 1 tzinfo = pytz.FixedOffset( sign * (int(timestamp[-4:-2]) * 60 + int(timestamp[-2:]))) return tzinfo.localize(dt)
def test_dumps_datetime_with_zone(self): """Testing dumps and loads of timezone-aware datetime types, as well as standalone timezone objects """ try: import pytz import tzlocal except ImportError: # These packages aren't available - can't test raise unittest.SkipTest( "pytz or tzlocal not available in this test run; skipping" "zone-aware DT tests.") local_tz = tzlocal.get_localzone() pytz_est = pytz.timezone("US/Eastern") pytz_pst = pytz.timezone("US/Pacific") pytz_utc = pytz.timezone("UTC") pytz_fixed = pytz.FixedOffset(-120) original_allow_pickle = morejson.CONFIG.get("allow_pickle", False) morejson.CONFIG["allow_pickle"] = True dicti = { 'datetime-no-tz': datetime.datetime.now(), 'datetime-with-utc': datetime.datetime.now(tz=pytz_utc), 'datetime-with-est': datetime.datetime.now(tz=pytz_est), 'datetime-with-tzlocal': datetime.datetime.now(tz=local_tz), 'datetime-with-pst': datetime.datetime.now(tz=pytz_pst), 'datetime-with-fixedoffset': datetime.datetime.now(tz=pytz_fixed), 'eastern-tzone': pytz_est, 'pacific-tzone': pytz_pst, 'array': [ 1, 2, 3, pytz_utc, pytz_est, local_tz, pytz_pst, pytz_fixed], 'string': 'trololo', 'null': None } out_str = morejson.dumps(dicti) actual_obj = morejson.loads(out_str) self.assertEqual(dicti, actual_obj) morejson.CONFIG["allow_pickle"] = original_allow_pickle
def test_pythonvalue(self): instance = builtins.gYearMonth() assert instance.pythonvalue('2001-10') == (2001, 10, None) assert instance.pythonvalue('2001-10+02:00') == (2001, 10, pytz.FixedOffset(120)) assert instance.pythonvalue('2001-10Z') == (2001, 10, pytz.utc) assert instance.pythonvalue('2001-10+00:00') == (2001, 10, pytz.utc) assert instance.pythonvalue('-2001-10') == (-2001, 10, None) assert instance.pythonvalue('-20001-10') == (-20001, 10, None) with pytest.raises(builtins.ParseError): assert instance.pythonvalue('10-10')
