Coverage for /var/srv/projects/api.amasfac.comuna18.com/tmp/venv/lib/python3.9/site-packages/pandas/tseries/frequencies.py: 18%

1from __future__ import annotations

3import warnings

5import numpy as np

7from pandas._libs.algos import unique_deltas

8from pandas._libs.tslibs import (

9 Timestamp,

10 get_unit_from_dtype,

11 periods_per_day,

12 tz_convert_from_utc,

13)

14from pandas._libs.tslibs.ccalendar import (

15 DAYS,

16 MONTH_ALIASES,

17 MONTH_NUMBERS,

18 MONTHS,

19 int_to_weekday,

20)

21from pandas._libs.tslibs.fields import (

22 build_field_sarray,

23 month_position_check,

24)

25from pandas._libs.tslibs.offsets import (

26 BaseOffset,

27 DateOffset,

28 Day,

29 _get_offset,

30 to_offset,

31)

32from pandas._libs.tslibs.parsing import get_rule_month

33from pandas._typing import npt

34from pandas.util._decorators import cache_readonly

35from pandas.util._exceptions import find_stack_level

37from pandas.core.dtypes.common import (

38 is_datetime64_dtype,

39 is_period_dtype,

40 is_timedelta64_dtype,

41)

42from pandas.core.dtypes.generic import (

43 ABCIndex,

44 ABCSeries,

45)

47from pandas.core.algorithms import unique

49# ---------------------------------------------------------------------

50# Offset names ("time rules") and related functions

52_offset_to_period_map = {

53 "WEEKDAY": "D",

54 "EOM": "M",

55 "BM": "M",

56 "BQS": "Q",

57 "QS": "Q",

58 "BQ": "Q",

59 "BA": "A",

60 "AS": "A",

61 "BAS": "A",

62 "MS": "M",

63 "D": "D",

64 "C": "C",

65 "B": "B",

66 "T": "T",

67 "S": "S",

68 "L": "L",

69 "U": "U",

70 "N": "N",

71 "H": "H",

72 "Q": "Q",

73 "A": "A",

74 "W": "W",

75 "M": "M",

76 "Y": "A",

77 "BY": "A",

78 "YS": "A",

79 "BYS": "A",

80}

82_need_suffix = ["QS", "BQ", "BQS", "YS", "AS", "BY", "BA", "BYS", "BAS"]

84for _prefix in _need_suffix:

85 for _m in MONTHS:

86 key = f"{_prefix}-{_m}"

87 _offset_to_period_map[key] = _offset_to_period_map[_prefix]

89for _prefix in ["A", "Q"]:

90 for _m in MONTHS:

91 _alias = f"{_prefix}-{_m}"

92 _offset_to_period_map[_alias] = _alias

94for _d in DAYS:

95 _offset_to_period_map[f"W-{_d}"] = f"W-{_d}"

98def get_period_alias(offset_str: str) -> str | None:

99 """

100 Alias to closest period strings BQ->Q etc.

101 """

102 return _offset_to_period_map.get(offset_str, None)

103

104

105def get_offset(name: str) -> BaseOffset:

106 """

107 Return DateOffset object associated with rule name.

108

109 .. deprecated:: 1.0.0

110

111 Examples

112 --------

113 get_offset('EOM') --> BMonthEnd(1)

114 """

115 warnings.warn(

116 "get_offset is deprecated and will be removed in a future version, "

117 "use to_offset instead.",

118 FutureWarning,

119 stacklevel=find_stack_level(),

120 )

121 return _get_offset(name)

122

123

124# ---------------------------------------------------------------------

125# Period codes

126

127

128def infer_freq(index, warn: bool = True) -> str | None:

129 """

130 Infer the most likely frequency given the input index.

131

132 Parameters

133 ----------

134 index : DatetimeIndex or TimedeltaIndex

135 If passed a Series will use the values of the series (NOT THE INDEX).

136 warn : bool, default True

137 .. deprecated:: 1.5.0

138

139 Returns

140 -------

141 str or None

142 None if no discernible frequency.

143

144 Raises

145 ------

146 TypeError

147 If the index is not datetime-like.

148 ValueError

149 If there are fewer than three values.

150

151 Examples

152 --------

153 >>> idx = pd.date_range(start='2020/12/01', end='2020/12/30', periods=30)

154 >>> pd.infer_freq(idx)

155 'D'

156 """

157 from pandas.core.api import (

158 DatetimeIndex,

159 Float64Index,

160 Index,

161 Int64Index,

162 )

163

164 if isinstance(index, ABCSeries):

165 values = index._values

166 if not (

167 is_datetime64_dtype(values)

168 or is_timedelta64_dtype(values)

169 or values.dtype == object

170 ):

171 raise TypeError(

172 "cannot infer freq from a non-convertible dtype "

173 f"on a Series of {index.dtype}"

174 )

175 index = values

176

177 inferer: _FrequencyInferer

178

179 if not hasattr(index, "dtype"):

180 pass

181 elif is_period_dtype(index.dtype):

182 raise TypeError(

183 "PeriodIndex given. Check the `freq` attribute "

184 "instead of using infer_freq."

185 )

186 elif is_timedelta64_dtype(index.dtype):

187 # Allow TimedeltaIndex and TimedeltaArray

188 inferer = _TimedeltaFrequencyInferer(index, warn=warn)

189 return inferer.get_freq()

190

191 if isinstance(index, Index) and not isinstance(index, DatetimeIndex):

192 if isinstance(index, (Int64Index, Float64Index)):

193 raise TypeError(

194 f"cannot infer freq from a non-convertible index type {type(index)}"

195 )

196 index = index._values

197

198 if not isinstance(index, DatetimeIndex):

199 index = DatetimeIndex(index)

200

201 inferer = _FrequencyInferer(index, warn=warn)

202 return inferer.get_freq()

203

204

205class _FrequencyInferer:

206 """

207 Not sure if I can avoid the state machine here

208 """

209

210 def __init__(self, index, warn: bool = True) -> None:

211 self.index = index

212 self.i8values = index.asi8

213

214 # For get_unit_from_dtype we need the dtype to the underlying ndarray,

215 # which for tz-aware is not the same as index.dtype

216 if isinstance(index, ABCIndex):

217 # error: Item "ndarray[Any, Any]" of "Union[ExtensionArray,

218 # ndarray[Any, Any]]" has no attribute "_ndarray"

219 self._reso = get_unit_from_dtype(

220 index._data._ndarray.dtype # type: ignore[union-attr]

221 )

222 else:

223 # otherwise we have DTA/TDA

224 self._reso = get_unit_from_dtype(index._ndarray.dtype)

225

226 # This moves the values, which are implicitly in UTC, to the

227 # the timezone so they are in local time

228 if hasattr(index, "tz"):

229 if index.tz is not None:

230 self.i8values = tz_convert_from_utc(self.i8values, index.tz)

231

232 if warn is not True:

233 warnings.warn(

234 "warn is deprecated (and never implemented) and "

235 "will be removed in a future version.",

236 FutureWarning,

237 stacklevel=find_stack_level(),

238 )

239 self.warn = warn

240

241 if len(index) < 3:

242 raise ValueError("Need at least 3 dates to infer frequency")

243

244 self.is_monotonic = (

245 self.index._is_monotonic_increasing or self.index._is_monotonic_decreasing

246 )

247

248 @cache_readonly

249 def deltas(self) -> npt.NDArray[np.int64]:

250 return unique_deltas(self.i8values)

251

252 @cache_readonly

253 def deltas_asi8(self) -> npt.NDArray[np.int64]:

254 # NB: we cannot use self.i8values here because we may have converted

255 # the tz in __init__

256 return unique_deltas(self.index.asi8)

257

258 @cache_readonly

259 def is_unique(self) -> bool:

260 return len(self.deltas) == 1

261

262 @cache_readonly

263 def is_unique_asi8(self) -> bool:

264 return len(self.deltas_asi8) == 1

265

266 def get_freq(self) -> str | None:

267 """

268 Find the appropriate frequency string to describe the inferred

269 frequency of self.i8values

270

271 Returns

272 -------

273 str or None

274 """

275 if not self.is_monotonic or not self.index._is_unique:

276 return None

277

278 delta = self.deltas[0]

279 ppd = periods_per_day(self._reso)

280 if delta and _is_multiple(delta, ppd):

281 return self._infer_daily_rule()

282

283 # Business hourly, maybe. 17: one day / 65: one weekend

284 if self.hour_deltas in ([1, 17], [1, 65], [1, 17, 65]):

285 return "BH"

286

287 # Possibly intraday frequency. Here we use the

288 # original .asi8 values as the modified values

289 # will not work around DST transitions. See #8772

290 if not self.is_unique_asi8:

291 return None

292

293 delta = self.deltas_asi8[0]

294 pph = ppd // 24

295 ppm = pph // 60

296 pps = ppm // 60

297 if _is_multiple(delta, pph):

298 # Hours

299 return _maybe_add_count("H", delta / pph)

300 elif _is_multiple(delta, ppm):

301 # Minutes

302 return _maybe_add_count("T", delta / ppm)

303 elif _is_multiple(delta, pps):

304 # Seconds

305 return _maybe_add_count("S", delta / pps)

306 elif _is_multiple(delta, (pps // 1000)):

307 # Milliseconds

308 return _maybe_add_count("L", delta / (pps // 1000))

309 elif _is_multiple(delta, (pps // 1_000_000)):

310 # Microseconds

311 return _maybe_add_count("U", delta / (pps // 1_000_000))

312 else:

313 # Nanoseconds

314 return _maybe_add_count("N", delta)

315

316 @cache_readonly

317 def day_deltas(self) -> list[int]:

318 ppd = periods_per_day(self._reso)

319 return [x / ppd for x in self.deltas]

320

321 @cache_readonly

322 def hour_deltas(self) -> list[int]:

323 pph = periods_per_day(self._reso) // 24

324 return [x / pph for x in self.deltas]

325

326 @cache_readonly

327 def fields(self) -> np.ndarray: # structured array of fields

328 return build_field_sarray(self.i8values, reso=self._reso)

329

330 @cache_readonly

331 def rep_stamp(self) -> Timestamp:

332 return Timestamp(self.i8values[0])

333

334 def month_position_check(self) -> str | None:

335 return month_position_check(self.fields, self.index.dayofweek)

336

337 @cache_readonly

338 def mdiffs(self) -> npt.NDArray[np.int64]:

339 nmonths = self.fields["Y"] * 12 + self.fields["M"]

340 return unique_deltas(nmonths.astype("i8"))

341

342 @cache_readonly

343 def ydiffs(self) -> npt.NDArray[np.int64]:

344 return unique_deltas(self.fields["Y"].astype("i8"))

345

346 def _infer_daily_rule(self) -> str | None:

347 annual_rule = self._get_annual_rule()

348 if annual_rule:

349 nyears = self.ydiffs[0]

350 month = MONTH_ALIASES[self.rep_stamp.month]

351 alias = f"{annual_rule}-{month}"

352 return _maybe_add_count(alias, nyears)

353

354 quarterly_rule = self._get_quarterly_rule()

355 if quarterly_rule:

356 nquarters = self.mdiffs[0] / 3

357 mod_dict = {0: 12, 2: 11, 1: 10}

358 month = MONTH_ALIASES[mod_dict[self.rep_stamp.month % 3]]

359 alias = f"{quarterly_rule}-{month}"

360 return _maybe_add_count(alias, nquarters)

361

362 monthly_rule = self._get_monthly_rule()

363 if monthly_rule:

364 return _maybe_add_count(monthly_rule, self.mdiffs[0])

365

366 if self.is_unique:

367 return self._get_daily_rule()

368

369 if self._is_business_daily():

370 return "B"

371

372 wom_rule = self._get_wom_rule()

373 if wom_rule:

374 return wom_rule

375

376 return None

377

378 def _get_daily_rule(self) -> str | None:

379 ppd = periods_per_day(self._reso)

380 days = self.deltas[0] / ppd

381 if days % 7 == 0:

382 # Weekly

383 wd = int_to_weekday[self.rep_stamp.weekday()]

384 alias = f"W-{wd}"

385 return _maybe_add_count(alias, days / 7)

386 else:

387 return _maybe_add_count("D", days)

388

389 def _get_annual_rule(self) -> str | None:

390 if len(self.ydiffs) > 1:

391 return None

392

393 if len(unique(self.fields["M"])) > 1:

394 return None

395

396 pos_check = self.month_position_check()

397

398 if pos_check is None:

399 return None

400 else:

401 return {"cs": "AS", "bs": "BAS", "ce": "A", "be": "BA"}.get(pos_check)

402

403 def _get_quarterly_rule(self) -> str | None:

404 if len(self.mdiffs) > 1:

405 return None

406

407 if not self.mdiffs[0] % 3 == 0:

408 return None

409

410 pos_check = self.month_position_check()

411

412 if pos_check is None:

413 return None

414 else:

415 return {"cs": "QS", "bs": "BQS", "ce": "Q", "be": "BQ"}.get(pos_check)

416

417 def _get_monthly_rule(self) -> str | None:

418 if len(self.mdiffs) > 1:

419 return None

420 pos_check = self.month_position_check()

421

422 if pos_check is None:

423 return None

424 else:

425 return {"cs": "MS", "bs": "BMS", "ce": "M", "be": "BM"}.get(pos_check)

426

427 def _is_business_daily(self) -> bool:

428 # quick check: cannot be business daily

429 if self.day_deltas != [1, 3]:

430 return False

431

432 # probably business daily, but need to confirm

433 first_weekday = self.index[0].weekday()

434 shifts = np.diff(self.index.asi8)

435 ppd = periods_per_day(self._reso)

436 shifts = np.floor_divide(shifts, ppd)

437 weekdays = np.mod(first_weekday + np.cumsum(shifts), 7)

438

439 return bool(

440 np.all(

441 ((weekdays == 0) & (shifts == 3))

442 | ((weekdays > 0) & (weekdays <= 4) & (shifts == 1))

443 )

444 )

445

446 def _get_wom_rule(self) -> str | None:

447 # FIXME: dont leave commented-out

448 # wdiffs = unique(np.diff(self.index.week))

449 # We also need -47, -49, -48 to catch index spanning year boundary

450 # if not lib.ismember(wdiffs, set([4, 5, -47, -49, -48])).all():

451 # return None

452

453 weekdays = unique(self.index.weekday)

454 if len(weekdays) > 1:

455 return None

456

457 week_of_months = unique((self.index.day - 1) // 7)

458 # Only attempt to infer up to WOM-4. See #9425

459 week_of_months = week_of_months[week_of_months < 4]

460 if len(week_of_months) == 0 or len(week_of_months) > 1:

461 return None

462

463 # get which week

464 week = week_of_months[0] + 1

465 wd = int_to_weekday[weekdays[0]]

466

467 return f"WOM-{week}{wd}"

468

469

470class _TimedeltaFrequencyInferer(_FrequencyInferer):

471 def _infer_daily_rule(self):

472 if self.is_unique:

473 return self._get_daily_rule()

474

475

476def _is_multiple(us, mult: int) -> bool:

477 return us % mult == 0

478

479

480def _maybe_add_count(base: str, count: float) -> str:

481 if count != 1:

482 assert count == int(count)

483 count = int(count)

484 return f"{count}{base}"

485 else:

486 return base

487

488

489# ----------------------------------------------------------------------

490# Frequency comparison

491

492

493def is_subperiod(source, target) -> bool:

494 """

495 Returns True if downsampling is possible between source and target

496 frequencies

497

498 Parameters

499 ----------

500 source : str or DateOffset

501 Frequency converting from

502 target : str or DateOffset

503 Frequency converting to

504

505 Returns

506 -------

507 bool

508 """

509

510 if target is None or source is None:

511 return False

512 source = _maybe_coerce_freq(source)

513 target = _maybe_coerce_freq(target)

514

515 if _is_annual(target):

516 if _is_quarterly(source):

517 return _quarter_months_conform(

518 get_rule_month(source), get_rule_month(target)

519 )

520 return source in {"D", "C", "B", "M", "H", "T", "S", "L", "U", "N"}

521 elif _is_quarterly(target):

522 return source in {"D", "C", "B", "M", "H", "T", "S", "L", "U", "N"}

523 elif _is_monthly(target):

524 return source in {"D", "C", "B", "H", "T", "S", "L", "U", "N"}

525 elif _is_weekly(target):

526 return source in {target, "D", "C", "B", "H", "T", "S", "L", "U", "N"}

527 elif target == "B":

528 return source in {"B", "H", "T", "S", "L", "U", "N"}

529 elif target == "C":

530 return source in {"C", "H", "T", "S", "L", "U", "N"}

531 elif target == "D":

532 return source in {"D", "H", "T", "S", "L", "U", "N"}

533 elif target == "H":

534 return source in {"H", "T", "S", "L", "U", "N"}

535 elif target == "T":

536 return source in {"T", "S", "L", "U", "N"}

537 elif target == "S":

538 return source in {"S", "L", "U", "N"}

539 elif target == "L":

540 return source in {"L", "U", "N"}

541 elif target == "U":

542 return source in {"U", "N"}

543 elif target == "N":

544 return source in {"N"}

545 else:

546 return False

547

548

549def is_superperiod(source, target) -> bool:

550 """

551 Returns True if upsampling is possible between source and target

552 frequencies

553

554 Parameters

555 ----------

556 source : str or DateOffset

557 Frequency converting from

558 target : str or DateOffset

559 Frequency converting to

560

561 Returns

562 -------

563 bool

564 """

565 if target is None or source is None:

566 return False

567 source = _maybe_coerce_freq(source)

568 target = _maybe_coerce_freq(target)

569

570 if _is_annual(source):

571 if _is_annual(target):

572 return get_rule_month(source) == get_rule_month(target)

573

574 if _is_quarterly(target):

575 smonth = get_rule_month(source)

576 tmonth = get_rule_month(target)

577 return _quarter_months_conform(smonth, tmonth)

578 return target in {"D", "C", "B", "M", "H", "T", "S", "L", "U", "N"}

579 elif _is_quarterly(source):

580 return target in {"D", "C", "B", "M", "H", "T", "S", "L", "U", "N"}

581 elif _is_monthly(source):

582 return target in {"D", "C", "B", "H", "T", "S", "L", "U", "N"}

583 elif _is_weekly(source):

584 return target in {source, "D", "C", "B", "H", "T", "S", "L", "U", "N"}

585 elif source == "B":

586 return target in {"D", "C", "B", "H", "T", "S", "L", "U", "N"}

587 elif source == "C":

588 return target in {"D", "C", "B", "H", "T", "S", "L", "U", "N"}

589 elif source == "D":

590 return target in {"D", "C", "B", "H", "T", "S", "L", "U", "N"}

591 elif source == "H":

592 return target in {"H", "T", "S", "L", "U", "N"}

593 elif source == "T":

594 return target in {"T", "S", "L", "U", "N"}

595 elif source == "S":

596 return target in {"S", "L", "U", "N"}

597 elif source == "L":

598 return target in {"L", "U", "N"}

599 elif source == "U":

600 return target in {"U", "N"}

601 elif source == "N":

602 return target in {"N"}

603 else:

604 return False

605

606

607def _maybe_coerce_freq(code) -> str:

608 """we might need to coerce a code to a rule_code

609 and uppercase it

610

611 Parameters

612 ----------

613 source : str or DateOffset

614 Frequency converting from

615

616 Returns

617 -------

618 str

619 """

620 assert code is not None

621 if isinstance(code, DateOffset):

622 code = code.rule_code

623 return code.upper()

624

625

626def _quarter_months_conform(source: str, target: str) -> bool:

627 snum = MONTH_NUMBERS[source]

628 tnum = MONTH_NUMBERS[target]

629 return snum % 3 == tnum % 3

630

631

632def _is_annual(rule: str) -> bool:

633 rule = rule.upper()

634 return rule == "A" or rule.startswith("A-")

635

636

637def _is_quarterly(rule: str) -> bool:

638 rule = rule.upper()

639 return rule == "Q" or rule.startswith("Q-") or rule.startswith("BQ")

640

641

642def _is_monthly(rule: str) -> bool:

643 rule = rule.upper()

644 return rule == "M" or rule == "BM"

645

646

647def _is_weekly(rule: str) -> bool:

648 rule = rule.upper()

649 return rule == "W" or rule.startswith("W-")

650

651

652__all__ = [

653 "Day",

654 "get_offset",

655 "get_period_alias",

656 "infer_freq",

657 "is_subperiod",

658 "is_superperiod",

659 "to_offset",

660]