def _handle_dividend_payable(self, trading_date):
"""handle dividend payable before trading
"""
to_delete_dividend = []
for order_book_id, dividend_info in six.iteritems(self.portfolio._dividend_info):
dividend_series_dict = dividend_info.dividend_series_dict
if pd.Timestamp(trading_date) == pd.Timestamp(dividend_series_dict['payable_date']):
dividend_per_share = dividend_series_dict["dividend_cash_before_tax"] / dividend_series_dict["round_lot"]
if dividend_per_share > 0 and dividend_info.quantity > 0:
dividend_cash = dividend_per_share * dividend_info.quantity
self.portfolio._dividend_receivable -= dividend_cash
self.portfolio._cash += dividend_cash
# user_log.info(_("get dividend {dividend} for {order_book_id}").format(
# dividend=dividend_cash,
# order_book_id=order_book_id,
# ))
to_delete_dividend.append(order_book_id)
for order_book_id in to_delete_dividend:
self.portfolio._dividend_info.pop(order_book_id, None)
python类Timestamp()的实例源码
def _handle_dividend_payable(self, trading_date):
"""handle dividend payable before trading
"""
to_delete_dividend = []
for order_book_id, dividend_info in six.iteritems(self.portfolio._dividend_info):
dividend_series_dict = dividend_info.dividend_series_dict
if pd.Timestamp(trading_date) == pd.Timestamp(dividend_series_dict['payable_date']):
dividend_per_share = dividend_series_dict["dividend_cash_before_tax"] / dividend_series_dict["round_lot"]
if dividend_per_share > 0 and dividend_info.quantity > 0:
dividend_cash = dividend_per_share * dividend_info.quantity
self.portfolio._dividend_receivable -= dividend_cash
self.portfolio._cash += dividend_cash
to_delete_dividend.append(order_book_id)
for order_book_id in to_delete_dividend:
self.portfolio._dividend_info.pop(order_book_id, None)
def is_valid_date(self, ignore_none=True):
def check_is_valid_date(func_name, value):
if ignore_none and value is None:
return None
if isinstance(value, (datetime.date, pd.Timestamp)):
return
if isinstance(value, six.string_types):
try:
v = parse_date(value)
except ValueError:
raise RQInvalidArgument(
_('function {}: invalid {} argument, expect a valid date, got {} (type: {})').format(
func_name, self._arg_name, value, type(value)
))
raise RQInvalidArgument(
_('function {}: invalid {} argument, expect a valid date, got {} (type: {})').format(
func_name, self._arg_name, value, type(value)
))
self._rules.append(check_is_valid_date)
return self
def get_previous_trading_date(date):
"""
?????????????
:param date: ????
:type date: `str` | `date` | `datetime` | `pandas.Timestamp`
:return: `datetime.date`
:example:
.. code-block:: python3
:linenos:
[In]get_previous_trading_date(date='2016-05-02')
[Out]
[datetime.date(2016, 4, 29)]
"""
return ExecutionContext.data_proxy.get_previous_trading_date(date)
def get_next_trading_date(date):
"""
????????????
:param date: ????
:type date: `str` | `date` | `datetime` | `pandas.Timestamp`
:return: `datetime.date`
:example:
.. code-block:: python3
:linenos:
[In]get_next_trading_date(date='2016-05-01')
[Out]
[datetime.date(2016, 5, 3)]
"""
return ExecutionContext.data_proxy.get_next_trading_date(date)
def is_valid_date(self, ignore_none=True):
def check_is_valid_date(func_name, value):
if ignore_none and value is None:
return None
if isinstance(value, (datetime.date, pd.Timestamp)):
return
if isinstance(value, six.string_types):
try:
v = parse_date(value)
return
except ValueError:
raise RQInvalidArgument(
_(u"function {}: invalid {} argument, expect a valid date, got {} (type: {})").format(
func_name, self._arg_name, value, type(value)
))
raise RQInvalidArgument(
_(u"function {}: invalid {} argument, expect a valid date, got {} (type: {})").format(
func_name, self._arg_name, value, type(value)
))
self._rules.append(check_is_valid_date)
return self
def get_trading_dates(start_date, end_date):
"""
???????????????????????????????????
:param start_date: ????
:type start_date: `str` | `date` | `datetime` | `pandas.Timestamp`
:param end_date: ????
:type end_date: `str` | `date` | `datetime` | `pandas.Timestamp`
:return: list[`datetime.date`]
:example:
.. code-block:: python3
:linenos:
[In]get_trading_dates(start_date='2016-05-05', end_date='20160505')
[Out]
[datetime.date(2016, 5, 5)]
"""
return Environment.get_instance().data_proxy.get_trading_dates(start_date, end_date)
def get_next_trading_date(date):
"""
????????????
:param date: ????
:type date: `str` | `date` | `datetime` | `pandas.Timestamp`
:return: `datetime.date`
:example:
.. code-block:: python3
:linenos:
[In]get_next_trading_date(date='2016-05-01')
[Out]
[datetime.date(2016, 5, 3)]
"""
return Environment.get_instance().data_proxy.get_next_trading_date(date)
def test_date_range_lower_freq():
cal = mcal.get_calendar("NYSE")
schedule = cal.schedule(pd.Timestamp('2017-09-05 20:00', tz='UTC'), pd.Timestamp('2017-10-23 20:00', tz='UTC'))
# cannot get date range of frequency lower than 1D
with pytest.raises(ValueError):
mcal.date_range(schedule, frequency='3D')
# instead get for 1D and convert to lower frequency
short = mcal.date_range(schedule, frequency='1D')
actual = mcal.convert_freq(short, '3D')
expected = pd.date_range('2017-09-05 20:00', '2017-10-23 20:00', freq='3D', tz='UTC')
assert_index_equal(actual, expected)
actual = mcal.convert_freq(short, '1W')
expected = pd.date_range('2017-09-05 20:00', '2017-10-23 20:00', freq='1W', tz='UTC')
assert_index_equal(actual, expected)
def test_2016_holidays():
# new years: jan 1
# mlk: jan 18
# presidents: feb 15
# good friday: mar 25
# mem day: may 30
# independence day: july 4
# labor day: sep 5
# thanksgiving day: nov 24
# christmas (observed): dec 26
# new years (observed): jan 2 2017
cfe = CFEExchangeCalendar()
good_dates = cfe.valid_days('2016-01-01', '2016-12-31')
for day in ["2016-01-01", "2016-01-18", "2016-02-15", "2016-03-25",
"2016-05-30", "2016-07-04", "2016-09-05", "2016-11-24",
"2016-12-26", "2017-01-02"]:
assert pd.Timestamp(day, tz='UTC') not in good_dates
def test_special_holidays():
# 9/11
# Sept 11, 12, 13, 14 2001
nyse = NYSEExchangeCalendar()
good_dates = nyse.valid_days('2001-01-01', '2016-12-31')
assert pd.Timestamp("9/11/2001") not in good_dates
assert pd.Timestamp("9/12/2001") not in good_dates
assert pd.Timestamp("9/13/2001") not in good_dates
assert pd.Timestamp("9/14/2001") not in good_dates
# Hurricane Sandy
# Oct 29, 30 2012
assert pd.Timestamp("10/29/2012") not in good_dates
assert pd.Timestamp("10/30/2012") not in good_dates
# various national days of mourning
# Gerald Ford - 1/2/2007
assert pd.Timestamp("1/2/2007") not in good_dates
# Ronald Reagan - 6/11/2004
assert pd.Timestamp("6/11/2004") not in good_dates
# Richard Nixon - 4/27/1994
assert pd.Timestamp("4/27/1994") not in good_dates
def test_2016_early_closes():
# mlk day: 2016-01-18
# presidents: 2016-02-15
# mem day: 2016-05-30
# july 4: 2016-07-04
# labor day: 2016-09-05
# thanksgiving: 2016-11-24
cme = CMEExchangeCalendar()
schedule = cme.schedule('2016-01-01', '2016-12-31')
early_closes = cme.early_closes(schedule).index
for date in ["2016-01-18", "2016-02-15", "2016-05-30", "2016-07-04",
"2016-09-05", "2016-11-24"]:
dt = pd.Timestamp(date, tz='UTC')
assert dt in early_closes
market_close = schedule.loc[dt].market_close
assert market_close.tz_convert(cme.tz).hour == 12
def test_clean_dates():
start, end = clean_dates('2016-12-01', '2016-12-31')
assert start == pd.Timestamp('2016-12-01')
assert end == pd.Timestamp('2016-12-31')
start, end = clean_dates('2016-12-01 12:00', '2016-12-31 12:00')
assert start == pd.Timestamp('2016-12-01')
assert end == pd.Timestamp('2016-12-31')
start, end = clean_dates(pd.Timestamp('2016-12-01', tz='America/Chicago'),
pd.Timestamp('2016-12-31', tz='America/New_York'))
assert start == pd.Timestamp('2016-12-01')
assert end == pd.Timestamp('2016-12-31')
start, end = clean_dates(pd.Timestamp('2016-12-01 09:31', tz='America/Chicago'),
pd.Timestamp('2016-12-31 16:00', tz='America/New_York'))
assert start == pd.Timestamp('2016-12-01')
assert end == pd.Timestamp('2016-12-31')
def test_schedule_w_times():
cal = FakeCalendar(time(12, 12), time(13, 13))
assert cal.open_time == time(12, 12)
assert cal.close_time == time(13, 13)
results = cal.schedule('2016-12-01', '2016-12-31')
assert len(results) == 21
expected = pd.Series({'market_open': pd.Timestamp('2016-12-01 04:12:00+0000', tz='UTC', freq='B'),
'market_close': pd.Timestamp('2016-12-01 05:13:00+0000', tz='UTC', freq='B')},
name=pd.Timestamp('2016-12-01'), index=['market_open', 'market_close'], dtype=object)
assert_series_equal(results.iloc[0], expected)
expected = pd.Series({'market_open': pd.Timestamp('2016-12-30 04:12:00+0000', tz='UTC', freq='B'),
'market_close': pd.Timestamp('2016-12-30 05:13:00+0000', tz='UTC', freq='B')},
name=pd.Timestamp('2016-12-30'), index=['market_open', 'market_close'], dtype=object)
assert_series_equal(results.iloc[-1], expected)
def test_special_closes():
cal = FakeCalendar()
results = cal.schedule('2012-07-01', '2012-07-06')
closes = results['market_close'].tolist()
# confirm that the day before July 4th is an 11:30 close not 11:49
assert pd.Timestamp('2012-07-02 11:49', tz='Asia/Ulaanbaatar').tz_convert('UTC') in closes
assert pd.Timestamp('2012-07-03 11:30', tz='Asia/Ulaanbaatar').tz_convert('UTC') in closes
assert pd.Timestamp('2012-07-04 11:49', tz='Asia/Ulaanbaatar').tz_convert('UTC') in closes
# early close first date
results = cal.schedule('2012-07-03', '2012-07-04')
actual = results['market_close'].tolist()
expected = [pd.Timestamp('2012-07-03 11:30', tz='Asia/Ulaanbaatar').tz_convert('UTC'),
pd.Timestamp('2012-07-04 11:49', tz='Asia/Ulaanbaatar').tz_convert('UTC')]
assert actual == expected
# early close last date
results = cal.schedule('2012-07-02', '2012-07-03')
actual = results['market_close'].tolist()
expected = [pd.Timestamp('2012-07-02 11:49', tz='Asia/Ulaanbaatar').tz_convert('UTC'),
pd.Timestamp('2012-07-03 11:30', tz='Asia/Ulaanbaatar').tz_convert('UTC')]
assert actual == expected
def test_special_closes_adhoc():
cal = FakeCalendar()
results = cal.schedule('2016-12-10', '2016-12-20')
closes = results['market_close'].tolist()
# confirm that 2016-12-14 is an 11:40 close not 11:49
assert pd.Timestamp('2016-12-13 11:49', tz='Asia/Ulaanbaatar').tz_convert('UTC') in closes
assert pd.Timestamp('2016-12-14 11:40', tz='Asia/Ulaanbaatar').tz_convert('UTC') in closes
assert pd.Timestamp('2016-12-15 11:49', tz='Asia/Ulaanbaatar').tz_convert('UTC') in closes
# now with the early close as end date
results = cal.schedule('2016-12-13', '2016-12-14')
closes = results['market_close'].tolist()
assert pd.Timestamp('2016-12-13 11:49', tz='Asia/Ulaanbaatar').tz_convert('UTC') in closes
assert pd.Timestamp('2016-12-14 11:40', tz='Asia/Ulaanbaatar').tz_convert('UTC') in closes
def test_open_at_time():
cal = FakeCalendar()
schedule = cal.schedule('2014-01-01', '2016-12-31')
# regular trading day
assert cal.open_at_time(schedule, pd.Timestamp('2014-07-02 03:40', tz='UTC')) is True
# early close
assert cal.open_at_time(schedule, pd.Timestamp('2014-07-03 03:40', tz='UTC')) is False
# holiday
assert cal.open_at_time(schedule, pd.Timestamp('2014-12-25 03:30', tz='UTC')) is False
# last bar of the day defaults to False
assert cal.open_at_time(schedule, pd.Timestamp('2016-09-07 11:49', tz='Asia/Ulaanbaatar')) is False
# last bar of the day is True if include_close is True
assert cal.open_at_time(schedule, pd.Timestamp('2016-09-07 11:49', tz='Asia/Ulaanbaatar'),
include_close=True) is True
def test_2016_early_closes():
# 2016 early closes
# mlk: 2016-01-18
# presidents: 2016-02-15
# mem day: 2016-05-30
# independence day: 2016-07-04
# labor: 2016-09-05
# thanksgiving: 2016-11-24
ice = ICEExchangeCalendar()
schedule = ice.schedule('2016-01-01', '2016-12-31')
early_closes = ice.early_closes(schedule)
for date in ["2016-01-18", "2016-02-15", "2016-05-30", "2016-07-04",
"2016-09-05", "2016-11-24"]:
dt = pd.Timestamp(date, tz='UTC')
assert dt in early_closes.index
market_close = schedule.loc[dt].market_close
# all ICE early closes are 1 pm local
assert market_close.tz_convert(ice.tz).hour == 13
def future_chain(self, root_symbol, as_of_date=None):
""" Look up a future chain with the specified parameters.
Parameters
----------
root_symbol : str
The root symbol of a future chain.
as_of_date : datetime.datetime or pandas.Timestamp or str, optional
Date at which the chain determination is rooted. I.e. the
existing contract whose notice date is first after this date is
the primary contract, etc.
Returns
-------
FutureChain
The future chain matching the specified parameters.
Raises
------
RootSymbolNotFound
If a future chain could not be found for the given root symbol.
"""
if as_of_date:
try:
as_of_date = pd.Timestamp(as_of_date, tz='UTC')
except ValueError:
raise UnsupportedDatetimeFormat(input=as_of_date,
method='future_chain')
return FutureChain(
asset_finder=self.asset_finder,
get_datetime=self.get_datetime,
root_symbol=root_symbol,
as_of_date=as_of_date
)
def set_symbol_lookup_date(self, dt):
"""
Set the date for which symbols will be resolved to their sids
(symbols may map to different firms or underlying assets at
different times)
"""
try:
self._symbol_lookup_date = pd.Timestamp(dt, tz='UTC')
except ValueError:
raise UnsupportedDatetimeFormat(input=dt,
method='set_symbol_lookup_date')
