using System;
using System.Collections.Generic;
using System.Globalization;
using System.IO;
using System.Net;
using System.Text.RegularExpressions;
using System.Threading;
using Zanetti.Data;
namespace Zanetti.DataSource.Specialized
{
internal class KabutanSource : DailyDataSource
{
private readonly Object _syncObject = new object();
private readonly List<int> _codes = new List<int>();
private Queue<int> _codeQueue;
private readonly List<int> _series = new List<int>();
private readonly Queue<FetchResult> _resultQueue = new Queue<FetchResult>();
private bool _terminate;
private Exception _exception;
private const int DaysAtOnce = 20; // 一度に取得する時系列の営業日数
private const int ThreadTimes = 1;
private class FetchResult
{
public enum Status
{
Success,
Failure,
Obsolete,
Retry,
}
public int Code;
public SortedDictionary<int, NewDailyData> Prices;
public Status ReturnStatus;
}
public KabutanSource(int[] dates) : base(dates)
{
foreach (AbstractBrand brand in Env.BrandCollection.Values)
{
var basic = brand as BasicBrand;
if (brand.Market == MarketType.B || brand.Market == MarketType.Custom ||
basic == null || basic.Obsolete)
continue;
_codes.Add(brand.Code);
}
}
public override int TotalStep
{
get { return (_codes.Count + 2) * ((_dates.Length + DaysAtOnce - 1) / DaysAtOnce); } // +2はNikkei225とTOPIX
}
public override void Run()
{
var threads = new Thread[ThreadTimes];
for (var i = 0; i < threads.Length; i++)
(threads[i] = new Thread(RunFetchPrices) { Name = "Fetch Thread " + i }).Start();
var dates = new List<int>(_dates);
try
{
do
{
// 日経平均の時系列データの存在を確認する。
var n = Math.Min(DaysAtOnce, dates.Count);
var original = dates.GetRange(0, n);
var nikkei225 = FetchPrices((int)BuiltInIndex.Nikkei225, original);
if (nikkei225.ReturnStatus != FetchResult.Status.Success)
throw new Exception(string.Format("株価の取得に失敗しました。時間を置いて再試行してください。: {0}～{1}",
original[0], original[original.Count - 1]));
dates.RemoveRange(0, n);
_series.Clear();
foreach (var date in original)
{
if (nikkei225.Prices[date].close == 0)
nikkei225.Prices.Remove(date);
else
_series.Add(date);
}
if (_series.Count == 0)
return;
UpdateDataFarm((int)BuiltInIndex.Nikkei225, _dates, nikkei225.Prices);
SendMessage(AsyncConst.WM_ASYNCPROCESS, (int)BuiltInIndex.Nikkei225, AsyncConst.LPARAM_PROGRESS_SUCCESSFUL);
_codeQueue = new Queue<int>(_codes);
_codeQueue.Enqueue((int)BuiltInIndex.TOPIX);
_codeQueue.Enqueue((int)BuiltInIndex.JASDAQ);
//_codeQueue.Enqueue((int)BuiltInIndex.TOSHO2BU);
//_codeQueue.Enqueue((int)BuiltInIndex.MOTHERS);
//_codeQueue.Enqueue((int)BuiltInIndex.JPYUSD);
//_codeQueue.Enqueue((int)BuiltInIndex.JPYEUR);
var retry = 0;
while (true)
{
int numCodes;
lock (_syncObject)
{
numCodes = _codeQueue.Count;
Monitor.PulseAll(_syncObject);
}
for (var i = 0; i < numCodes; i++)
{
FetchResult result;
lock (_resultQueue)
{
while (_resultQueue.Count == 0 && _exception == null)
Monitor.Wait(_resultQueue);
if (_exception != null)
throw _exception;
result = _resultQueue.Dequeue();
}
switch (result.ReturnStatus)
{
case FetchResult.Status.Failure:
case FetchResult.Status.Obsolete:
continue;
case FetchResult.Status.Retry:
lock (_codeQueue)
{
_codeQueue.Enqueue(result.Code);
}
continue;
}
UpdateDataFarm(result.Code, _dates, result.Prices);
SendMessage(AsyncConst.WM_ASYNCPROCESS, result.Code, AsyncConst.LPARAM_PROGRESS_SUCCESSFUL);
}
if (_codeQueue.Count == 0)
break;
if (retry++ == 10)
throw new Exception(string.Format("株価の取得に失敗しました。時間を置いて再試行してください。: {0}～{1}",
_series[0], _series[_series.Count - 1]));
Thread.Sleep(10000);
}
} while (dates.Count > 0);
}
finally
{
lock (_syncObject)
{
_terminate = true;
Monitor.PulseAll(_syncObject);
}
foreach (var thread in threads)
thread.Join();
}
}
public void UpdateDataFarm(int code, int[] _dates, SortedDictionary<int, NewDailyData> prices)
{
var farm = (DailyDataFarm)Env.BrandCollection.FindBrand(code).CreateDailyFarm(prices.Count);
var empty = farm.IsEmpty;
var skip = true;
foreach (var pair in prices)
{
if (!ChkDates(_dates, pair.Key)) continue;//差し替え その１で追加
if (empty && skip && pair.Value.volume == 0)
continue;
skip = false;
farm.UpdateDataFarm(pair.Key, pair.Value);
}
farm.Save(Util.GetDailyDataFileName(code));
}
private bool ChkDates(int[] _d, int date)//差し替え その１で追加
{
for(int i = 0;i<_d.Length;i++)
{
if (_d[i] == date)
return true;
}
return false;
}
private void RunFetchPrices()
{
var code = 0;
try
{
while (true)
{
lock (_syncObject)
{
while ((_codeQueue == null || _codeQueue.Count == 0) && !_terminate)
Monitor.Wait(_syncObject);
if (_terminate || _codeQueue == null)
return;
code = _codeQueue.Dequeue();
}
var result = FetchPrices(code, _series);
lock (_resultQueue)
{
_resultQueue.Enqueue(result);
Monitor.Pulse(_resultQueue);
}
}
}
catch (Exception e)
{
lock (_resultQueue)
{
_exception = new Exception(string.Format("{0}: {1} {2}", e.Message, code, _series[0]), e);
Monitor.Pulse(_resultQueue);
}
}
}
private FetchResult FetchPrices(int code, IList<int> dates)
{
string page;
var status = GetPage(code, Util.IntToDate(dates[0]), Util.IntToDate(dates[dates.Count - 1]), out page);
if (status == FetchResult.Status.Failure || status == FetchResult.Status.Retry)
return new FetchResult { Code = code, ReturnStatus = status };
return ParsePage(code, page, dates);
}
private FetchResult.Status GetPage(int code, DateTime begin, DateTime end, out string page)
{
if (code == (int)BuiltInIndex.Nikkei225)
code = 0;
else if (code == (int)BuiltInIndex.TOPIX)
code = 10;
else if (code == (int)BuiltInIndex.JASDAQ)
code = 102;
//else if (code == (int)BuiltInIndex.TOSHO2BU)
// code = 11;
//else if (code == (int)BuiltInIndex.MOTHERS)
// code = 12;
//else if (code == (int)BuiltInIndex.JPYUSD)
// code = 950;
//else if (code == (int)BuiltInIndex.JPYEUR)
// code = 951;
var url = string.Format(
"https://kabutan.jp/stock/kabuka?code={0:D4}", code);
page = null;
try
{
using (var reader = new StreamReader(Util.HttpDownload(url)))
page = reader.ReadToEnd();
}
catch (WebException e)
{
switch (e.Status)
{
case WebExceptionStatus.ProtocolError:
switch (((HttpWebResponse)e.Response).StatusCode)
{
case (HttpStatusCode)999:
case HttpStatusCode.InternalServerError:
case HttpStatusCode.BadGateway:
return FetchResult.Status.Retry;
}
throw;
case WebExceptionStatus.Timeout:
case WebExceptionStatus.ConnectionClosed:
case WebExceptionStatus.ReceiveFailure:
case WebExceptionStatus.ConnectFailure:
return FetchResult.Status.Retry;
default:
throw;
}
}
Thread.Sleep(1000);
return FetchResult.Status.Success;
}
private FetchResult ParsePage(int code, string buf, IEnumerable<int> dates)
{
buf = buf.Replace("－", "0");//差し替え その１で追加
var valid = new Regex(
@"<td style=""text-align:center;"">(?<year>\d{2})/(?<month>\d?\d)/(?<day>\d?\d)</td>\r\n" +
"<td>(?<open>[0-9,.]+)</td>\r\n<td>(?<high>[0-9,.]+)</td>\r\n<td>(?<low>[0-9,.]+)</td>\r\n<td>(?<close>[0-9,.]+)</td>\r\n" +
"<td>.*</td>\r\n<td>.*</td>\r\n<td>(?<volume>[0-9,]+)</td>");
var invalid = new Regex("該当する期間のデータはありません。<br>期間をご確認ください。");
var obs = new Regex("該当する銘柄はありません。<br>再度銘柄（コード）を入力し、「表示」ボタンを押してください。");
var empty = new Regex("<dl class=\"stocksInfo\">\n<dt></dt><dd class=\"category yjSb\"></dd>");
if (buf == null)
return null;
var dict = new SortedDictionary<int, NewDailyData>();
var matches = valid.Matches(buf);
if (matches.Count == 0)
{
if (obs.Match(buf).Success || empty.Match(buf).Success) // 上場廃止(銘柄データが空のこともある)
return new FetchResult { ReturnStatus = FetchResult.Status.Obsolete };
if (!invalid.Match(buf).Success)
throw new Exception("ページから株価を取得できません。");
// ここに到達するのは出来高がないか株価が用意されていない場合
}
try
{
var shift = IsIndex(code) ? 100 : 10; // 指数は100倍、株式は10倍で記録する
const NumberStyles s = NumberStyles.AllowDecimalPoint | NumberStyles.AllowThousands;
foreach (Match m in matches)
{
var date = new DateTime(int.Parse(m.Groups["year"].Value) + 2000,
int.Parse(m.Groups["month"].Value),
int.Parse(m.Groups["day"].Value));
dict[Util.DateToInt(date)] = new NewDailyData
{
open = (int)(double.Parse(m.Groups["open"].Value, s) * shift),
high = (int)(double.Parse(m.Groups["high"].Value, s) * shift),
low = (int)(double.Parse(m.Groups["low"].Value, s) * shift),
close = (int)(double.Parse(m.Groups["close"].Value, s) * shift),
volume = m.Groups["volume"].Value == "" ? 0 : (int)double.Parse(m.Groups["volume"].Value, s)
};
}
}
catch (FormatException e)
{
throw new Exception("ページから株価を取得できません。", e);
}
// 出来高がない日の株価データがないので値が0のデータを補う。
foreach (var date in dates)
{
if (!dict.ContainsKey(date))
dict[date] = new NewDailyData();
}
return new FetchResult { Code = code, Prices = dict, ReturnStatus = FetchResult.Status.Success };
}
private bool IsIndex(int code)
{
return code == (int)BuiltInIndex.Nikkei225 ||
code == (int)BuiltInIndex.TOPIX ||
code == (int)BuiltInIndex.JASDAQ;// ||
//code == (int)BuiltInIndex.MOTHERS ||
//code == (int)BuiltInIndex.TOSHO2BU ||
//code == (int)BuiltInIndex.JPYUSD ||
//code == (int)BuiltInIndex.JPYEUR;
}
}
}